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

如何用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实现机器学习算法带来帮助。如果在尝试过程中遇到什么问题,欢迎在评论区提问。

上一篇:年终盘点丨细数2017云栖社区20大热点话题(附100+话题清单)


下一篇:开源分布式文件系统比较