如何用Apache Spark和LightGBM构建机器学习模型来预测信用卡欺诈
编译：抚月，阿里巴巴计算平台事业部 EMR 高级工程师，Apache HDFS Committer，目前从事开源大数据存储和优化方面的工作。
任何一家发行信用卡的金融机构，都要做防欺诈检测。他们平台上每天不断的成千上万的交易记录，他们用这些大规模的数据集来做检测。虽然在个人电脑上，在Jupyter notebook上面对小量的静态数据集做机器学习的模型训练，相对比较简单。 但是在金融机构的真实环境下，大量的交易数据存放在Hadoop或者数据湖里，这时部署机器学习模型就存在巨大挑战。这篇文章，我将向你们展示如何在Apache Spark环境下，利用LightGBM来构建和训练模型。 LightGBM被认为是Gradient Boosting算法的高效实现，在数据科学领域被广泛使用。
Apache Spark是一个in-memory的集群化的运行环境，用来存储和处理数据，比起从(单机）SSD或磁盘访问数据会快上千倍。假设数据集存在于Hadoop，S3，数据库或本地文件，第一步需要让Spark集群可以访问这些数据（将数据转为Spark dataframe)。在这个例子中，我将Machine Learning Group of ULB的数据，以CSV格式加载到Spark dataframe中。 Spark集群是6gb内存，部署了Databricks社区版。下面是用Spark加载数据的代码：
# File location and type
file_location = "/FileStore/tables/creditcard.csv"
file_type = "csv"
# CSV options
infer_schema = "true"
first_row_is_header = "true"
delimiter = ","
df = spark.read.format(file_type) \
.option("inferSchema", infer_schema) \
.option("header", first_row_is_header) \
.option("sep", delimiter) \
.load(file_location)数据集包含了300,000条记录，31个变量，是关于欧洲信用卡持有者的交易记录。其中28个变量是数值型，是对一些未披露的原始参数进行主成分分析（PCA）得到的。剩下的3个变量是交易数量，交易时间（秒，相对于第一次交易），交易标签（表示是否真实或诈骗）。虽然数据集比较小，但是我们选择了Apache Spark来做训练，因此相同的代码同样适用于大规模的数据集。
数据集中，标记为真实的数据， 和标记为诈骗的数据相比，数量严重不平衡。其中只有492个样本是跟标记为诈骗的交易有关，比起整个数据集来说量很小。你可以访问这个数据集。
下一步，是从dataframe中，选择我们想要作为输入变量和目标变量的列。当构建生产级的ML模型时，比较好的做法是做一些数据变换，比如用LabelEncoder或者OneHotEncoder的方式，将类目列转为数值列，作为整个pipeline当中的几个stage。这个pipeline可以用来变换训练、验证、或者测试数据，而不会将测试用例暴露给训练数据（当我们做standardized scaling的时候）。而且代码也好维护。可以参考Spark的pipeline接口文档。
feature_cols = ["V" + str(i) for i in range(1,29)] + ["Amount"]
assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
stages = [assembler]下一步，我们添加一个LightGBMClassifier实例到pipeline。LightGBMClassifier这个类包含在MMLSpark库中，是由Microsoft Azure团队维护的一个库。按照这个步骤，可以将其添加到集群中。我通过maven依赖引入了mmlspark这个包，使用版本0.17。然后，可以通过Python import LightGBMClassifier。
使用LightGBM的时候，对超参数(hyperparameters)进行调优非常重要，比如叶子数量、最大深度、迭代次数等。对相同数据训练出来的模型，性能差异可能会因为超参数的值的不同而非常大。
通常，这需要进行一个交叉验证实验，用一些超参数值空间搜索策略，比如全网格搜索、随机搜索、贝叶斯优化，或者树状结构Parzen估计方法（TPE）来找到超参数的最优值，从而使得验证数据集上模型的性能最大化。通过最小化一些预定义的错误量，比如二进制错误；或者最大化一些分数，比如ROC曲线下的面积或者F1分数。
根据我们想搜索的参数值空间的大小，以及我们期望得到的模型的性能的不同，这些搜索可能会跑很长时间。在这种情况下，我决定调优7个LightGBM模型的超参数。大多数参数是实值，意味着搜索的参数空间是非常深的。
我用了Kaggle执行环境里面的Hyperopt库。为了在有限的时间内搜索超参数空间，评估不同的超参数值的组合，我使用了一些优化策略来找到超参数的近乎最优值，只用了较少的几个评估回合。我用树状结构Parzen估计方法，在200次的超参数组合的评估之后， 我找到了超参数的最优值。 notebook里面，是我在Spark里训练这个模型时，用来确定超参数的代码。
我用Databricks的社区版，花了2个小时。如果你可以有更多的时间访问Spark集群，你也可以用Spark里面的ParamGridBuilder和CrossValidator这2个类来搜索和评估超参数值。下面是实例代码。
下面，是我在Kaggle的python环境中进行模型调优，得到的大多数最优超参数值，以及Spark例子中的lambda L2正则化参数。注意，我设置了LightGBMClassifier的一个值isUnbalance=True，从而可以处理之前提到的数据集不平衡的问题。
best_params = {
'bagging_fraction': 0.8,
'bagging_freq': 1,
'eval_metric': 'binary_error',
'feature_fraction': 0.944714847210862,
'lambda_l1': 1.0,
'lambda_l2': 45.0,
'learning_rate': 0.1,
'loss_function': 'binary_error',
'max_bin': 60,
'max_depth': 58,
'metric': 'binary_error',
'num_iterations': 379,
'num_leaves': 850,
'objective': 'binary',
'random_state': 7,
'verbose': None
}
lgb = LightGBMClassifier(
learningRate=0.1,
earlyStoppingRound=100,
featuresCol='features',
labelCol='label',
isUnbalance=True,
baggingFraction=best_params["bagging_fraction"],
baggingFreq=1,
featureFraction=best_params["feature_fraction"],
lambdaL1=best_params["lambda_l1"],
# lambdaL2=best_params["lambda_l2"],
maxBin=best_params["max_bin"],
maxDepth=best_params["max_depth"],
numIterations=best_params["num_iterations"],
numLeaves=best_params["num_leaves"],
objective="binary",
baggingSeed=7
)
paramGrid = ParamGridBuilder().addGrid(
lgb.lambdaL2, list(np.arange(1.0, 101.0, 10.0))
).build()
evaluator = BinaryClassificationEvaluator(labelCol="label",metricName="areaUnderROC")
crossValidator = CrossValidator(estimator=lgb,
estimatorParamMaps=paramGrid,
evaluator=evaluator,
numFolds=2)
stages += [crossValidator]
pipelineModel = Pipeline(stages=stages)下一步是将数据集分为训练集和测试集。我们会用训练集去拟合我们刚才创建的pipeline（调用Spark pipeline的fit方法），其中包含了特征装配和模型训练。然后我们用这个pipeline去变换测试集，来生成预测结果。
train, test = df.randomSplit([0.8, 0.2], seed=7)
model = pipelineModel.fit(train)
preds = model.transform(test)一旦我们从测试数据集拿到预测结果，我们可以用它们衡量我们模型的性能。Spark提供了BinaryClassificationEvaluator，可以用来计算ROC曲线的面积。
为了计算其它相关的指标，比如精度、召回、F1分数，我们可以利用测试集的预测标签和实际标签。
binaryEvaluator = BinaryClassificationEvaluator(labelCol="label")
print ("Test Area Under ROC: " + str(binaryEvaluator.evaluate(preds, {binaryEvaluator.metricName: "areaUnderROC"})))
#True positives
tp = preds[(preds.label == 1) & (preds.prediction == 1)].count()
#True negatives
tn = preds[(preds.label == 0) & (preds.prediction == 0)].count()
#False positives
fp = preds[(preds.label == 0) & (preds.prediction == 1)].count()
#Falsel negatives
fn = preds[(preds.label == 1) & (preds.prediction == 0)].count()
print ("True Positives:", tp)
print ("True Negatives:", tn)
print ("False Positives:", fp)
print ("False Negatives:", fn)
print ("Total", preds.count())
r = float(tp)/(tp + fn)
print ("recall", r)
p = float(tp) / (tp + fp)
print ("precision", p)
f1 = 2 * p * r /(p + r)
print ("f1", f1)在这个例子里，AUC-ROC分数是0.93， F1分数是0.70. 在Kaggle notebook里，在训练之前，我还用了SMOTE来平衡数据集，结果AUC-ROC分数是0.98， F1分数是0.80。我对超参数值的组合做了200次的评估。如果我们想进一步提高模型的性能，那么下一步显然是在Spark里改进SMOTE，在拟合流水线之前先平衡一下训练集。我们还可以更深入地搜索超参数空间，通过执行更多次的超参数值组合的评估的方式。
除了在Spark上训练一个高性能的LightGBM模型，数据科学家遇到的另一个巨大挑战是管理这样一个生命周期----准备数据、选择模型、训练、调优、保存最优参数值、部署训练好的模型、通过API访问输出的预测值。MLFlow是一个开源的解决方案，可以解决这些问题。我建议感兴趣的读者去学习它。我可以写一篇关于如何如何将它整合进工作流里面。就像我上面展示的那样。
你可以参考上面的那些步骤，我放在了Github repo上。这是另一个notebook我用来进行参数调优实验。
我希望这篇文章对你开始在Apache Spark上用LightGBMClassifier实现机器学习算法带来帮助。如果在尝试过程中遇到什么问题，欢迎在评论区提问。