如何用Apache Spark和LightGBM构建机器学习模型来预测信用卡欺诈

原文链接 https://towardsdatascience.com/how-to-perform-credit-card-fraud-detection-at-large-scale-with-apache-spark-and-lightgbm-61a982de8672

编译：抚月，阿里巴巴计算平台事业部 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实现机器学习算法带来帮助。如果在尝试过程中遇到什么问题，欢迎在评论区提问。