如何使用Apache Spark MLlib预测电信客户流失

Spark MLLib是一个用于在海量数据集上执行机器学习和相关任务的库。使用MLlib，可以对十亿个观测值进行机器学习模型的拟合，可能只需要几行代码并利用数百台机器就能达到。MLlib大大简化了模型开发过程。

在本文中，我们将使用MLlib来拟合机器学习模型，该模型可以预测电信公司的哪些客户可能会停止使用他们的服务。流失预测，是电信行业和许多基于订阅行业中最常见的机器学习应用之一。

我们将使用Python编程语言来执行我们的分析和建模，并且我们将为该任务使用各种相关的工具。为了加载和处理数据，我们将使用Spark的DataFrames API。为了执行特征工程，模型拟合和模型评估，我们将使用Spark的ML Pipelines API。（MLlib的核心是在Cloudera Enterprise 5.5支持的CDH 5.5内部发布的，但直到未来发行版才会支持ML管道。）

本文基于我们在Strata + Hadoop World Singapore 2015 上的“Data Science for Telecom”教程中介绍的材料。完整的源代码和输出可在IPython笔记本中找到。该仓库还包含一个脚本，显示如何在CDH群集上启动具有所需依赖关系的IPython笔记本。

使用Spark DataFrames加载数据

我们将使我们的模型拟合由SGI托管的UC Irvine机器学习库提供的流失数据集。在这个数据集中，每条记录包含与单个订户对应的信息，以及该订户是否继续使用该服务。

该数据集仅包含5,000个观察者，即订阅者，比Spark能够处理的要小很多个数量级，但使用这种大小的数据可以轻松地在笔记本电脑上试用这些工具。

数据订阅中的全部字段是：

state 国家

account length 账户长度

area code 区号电话号码

international plan 国际计划

voice mail plan 语音邮件计划

number vmail message 号码vmail消息

total day minutes 总日分钟数

total day calls 总日通话数

total day charge 总日总费用

total eve minutes 总共前夕分钟数

total eve calls 总共前夕通话数

total eve charge 总前夕费用数

total night minutes 总夜间分钟数

total night calls 总夜间通话数

total night charge 总夜间费用数

total intl minutes 总国际分钟数

total intl calls 总国际通话数

total intl charge 总国际收费数

number customer service calls 号码客户服务电话数

churned 流失

最后一个字段“churned(流失)”，一个可以取值为“真”或“假”的分类变量，这是我们想要预测的标签。其余的字段将进行公平的竞赛，来产生独立变量，这些变量与模型结合使用用来生成预测值。

要将这些数据加载到Spark DataFrame中，我们只需告诉Spark每个字段的类型。我们使用Spark Spark项目之外的spark-csv包来解释CSV格式的数据：

from pyspark.sql import SQLContext
from pyspark.sql.types import *

sqlContext = SQLContext(sc)
schema = StructType([ \
    StructField("state", StringType(), True), \
    StructField("account_length", DoubleType(), True), \
    StructField("area_code", StringType(), True), \
    StructField("phone_number", StringType(), True), \
    StructField("intl_plan", StringType(), True), \
    StructField("voice_mail_plan", StringType(), True), \
    StructField("number_vmail_messages", DoubleType(), True), \
    StructField("total_day_minutes", DoubleType(), True), \
    StructField("total_day_calls", DoubleType(), True), \
    StructField("total_day_charge", DoubleType(), True), \
    StructField("total_eve_minutes", DoubleType(), True), \
    StructField("total_eve_calls", DoubleType(), True), \
    StructField("total_eve_charge", DoubleType(), True), \
    StructField("total_night_minutes", DoubleType(), True), \
    StructField("total_night_calls", DoubleType(), True), \
    StructField("total_night_charge", DoubleType(), True), \
    StructField("total_intl_minutes", DoubleType(), True), \
    StructField("total_intl_calls", DoubleType(), True), \
    StructField("total_intl_charge", DoubleType(), True), \
    StructField("number_customer_service_calls", DoubleType(), True), \
    StructField("churned", StringType(), True)])

churn_data = sqlContext.read \
    .format('com.databricks.spark.csv') \
    .load('churn.all', schema = schema)

拟合机器学习模型

MLlib提供了一系列算法，这些算法在大数据集上进行了拟合，并且可以进行相关统计处理。特别是我们将要使用的ML Pipelines API，它是一个这样的框架，可以用于在DataFrame中获取数据，应用转换来提取特征，并将提取的数据特征提供给机器学习算法。我们将使用MLlib来训练和评估一个可以预测用户是否可能流失的随机森林模型。

监督机器学习模型的开发和评估的广泛流程如下所示：

监督机器学习模型开发和评估流程

流程从数据集开始，数据集由可能具有多种类型的列组成。在我们的例子中，数据集是churn_data，这是我们在上面的部分中创建的。然后我们对这些数据进行特征提取，将其转换为一组特征向量和标签。特征向量是浮点数值的数组，表示我们的模型可用于进行预测的自变量。标签是代表我们的机器学习算法试图预测的因变量的单个浮点值。在我们这样的二元分类问题中，我们使用0.0和1.0来表示两种可能的预测结果。在我们的例子中，0.0意味着“不会流失”，1.0意味着“会流失”。

特征提取是指我们可能会关注从输入数据中产生特征向量和标签的一系列可能的转换。在我们的例子中，我们会将输入数据中用字符串表示的类型变量，如intl_plan转化为数字，并index（索引）它们。

我们将会选择列的一个子集。例如，我们不期待phone_number可能是一个非常有用的特征，所以我们可以将它从模型中删除，但是total_day_calls很可能是一个非常有用的特征，所以我们希望将其包含在内。我们通过定义两个阶段：StringIndexer和VectorAssembler，将这些转换步骤纳入我们的管道。

from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import VectorAssembler

label_indexer = StringIndexer(inputCol = 'churned', outputCol = 'label')
plan_indexer = StringIndexer(inputCol = 'intl_plan', outputCol = 'intl_plan_indexed')

reduced_numeric_cols = ["account_length", "number_vmail_messages", "total_day_calls",
                        "total_day_charge", "total_eve_calls", "total_eve_charge",
                        "total_night_calls", "total_intl_calls", "total_intl_charge"]

assembler = VectorAssembler(
    inputCols = ['intl_plan_indexed'] + reduced_numeric_cols,
    outputCol = 'features')

提取特征后，我们的下一步是将我们的数据集分割为train（训练集）和test（测试集）。机器学习算法将使用训练集来拟合模型。测试集将用于评估模型：

(train, test) = churn_data.randomSplit([0.7, 0.3])

现在我们可以组装好我们的管道并最终拟合模型。定义管道的一个优点是，你将了解到相同的代码正在应用于特征提取阶段。使用MLlib，这里只需要几行简短的代码！

from pyspark.ml import Pipeline
from pyspark.ml.classification import RandomForestClassifier

classifier = RandomForestClassifier(labelCol = 'label', featuresCol = 'features')
pipeline = Pipeline(stages=[plan_indexer, label_indexer, assembler, classifier])
model = pipeline.fit(train)

验证模型

我们怎么知道我们训练的模型是否是一个好模型？我们可以证明它产生的预测比随机猜测更好吗？对于二元分类模型，有用的评估指标是ROC曲线下的面积。通过采用二值分类预测器来产生ROC曲线，该预测器使用阈值来给连续预测值的定标签。当你改变模型的阈值时，会出现两种极端的情况，一种情况是真阳性概率（TPR）和假阳性概率（FPR）同时为0，因为所有内容都标注为“未流失”，另外一种情况是TPR和FPR两者都为1，因为一切都被贴上了“流失”的标签。

一个随机的预测器会将一半客户标记为流失，另一半客户标记为非流失，将会产生一条直对角线的ROC曲线。这条线将单位正方形切割成两个大小相等的三角形，因此曲线下方的面积为0.5。0.5的AUROC(AreaUnderROC，ROC曲线下面积)值意味着你的预测器在两个类别之间的区分性并不比随机猜测更好。值越接近1.0，预测越好。低于0.5的值表示我们可以通过反转它给我们的答案来使我们的模型产生更好的预测。

MLlib也使计算AUROC非常容易。如果我们要基于我们所有的数据计算ROC曲线，我们的分类评估指标就会过于乐观，因为我们会用我们训练的数据来评估一个模型。我们只用我们的测试集对模型进行评估，以避免模型评估指标（如AUROC）过于乐观，以及帮助我​​们避免过度拟合。

from pyspark.ml.evaluation import BinaryClassificationEvaluator

predictions = model.transform(test)
evaluator = BinaryClassificationEvaluator()
auroc = evaluator.evaluate(predictions, {evaluator.metricName: "areaUnderROC"})

在这种情况下，我们产生的曲线下面的面积是大于0.8，表明模型的结果相当好，并且肯定比随机猜测好。

结论

这篇文章仅提供了MLlib可能用例的一个例子。有关机器学习和Spark一般情况的更多示例，请参阅此列表。

Juliet Hougland是Cloudera的数据科学家，也是Sparkling Pandas项目的贡献者/提交者/维护者。

Sandy Ryza是Cloudera的数据科学家，也是Apache Spark和Apache Hadoop项目的提交者。他是 O'Reilly Media 的《高级分析与Spark》 的合着者。