亚马逊AWS官方博客
使用 Spark 与 Apache Arrow 同时训练多个机器学习模型
作者：Itzik Jan, Perion Network公司高级软件开发员 ; Avi Ben Yossef, Perion Network公司AI团队负责人
 |
 |
 |
在事件层级处理广告宣传数据时，我们往往瞬间被数十亿日常事件彻底吞没。
Python的经典数据与各类机器学习库——包括Pandas、Scikit-learn以及XGBoost等——在设计之初并没有考虑到如此恐怖的数据总量。这意味着我们只能做出痛苦的折衷，要么选择极简采样，要么放弃一部分重要功能。
在本文中，我们将重点介绍一项用例，其通过Pandas、Scikit-learn、PyArrow以及PySpark分发多套模型，借此改善模型的训练与测试性能、以及与业务目标相关的准确率指标。
作为AWS合作伙伴网络（APN）的指定技术合作方，Perion Network公司致力于管理并匹配互联网与移动广告领域的供需关系。
Perion的人工智能（AI）团队提供实时性能优化服务、推荐系统以及供需预测工具。这一切将帮助客户解析海量广告数据，并由此建立起复杂的机器学习建模流程。
背景介绍
我们的用例旨在构建起能够处理数十亿条记录的机器学习模型管道，借此预测并优化广告投放引擎所对应的点击率（CTR）与视频完成率（VCR）。我们使用PySpark进行数据准备，并使用Scikit-learn实现ML模型训练。
在第一轮模型训练迭代中，我们得到的准确率与典型线面积（AUC）都不够理想。训练耗费的时间很长，模型体积也相当夸张。此外，当我们尝试以实验方式将其部署到生产环境中时，由于日志数据量太少，广告产品的实际性能表现也不尽人意。
在对历史日志（其中包含各种广告产品的相关数据）进行分析之后，我们发现每种产品在行为模式、行数、展示次数以及事件效果（例如点击率与视频完成率等）方面存在着巨大差异。
我们还尝试以多种不同类型的不平衡数据策略，对各类广告产品的事件数量差异加以处理。我们发现虽然准确率有所提高，但仍然达不到预期水平。
最后，我们得出结论，即可以将数据拆分成多个较小的数据集（每个数据集包含单一广告产品的相关值）以改善模型性能。以此为基础，我们可以独立训练各个数据集，为每款广告产品创建独立的模型，并借此实现不平衡数据处理策略。
相较于生产统一的模型，我们对各类广告产品所对应的不同模型进行分别计算。好消息来了，每种新模型都带来了高于原始模型的AUC水平——原始模型的AUC约为0.63，而新模型则普遍介于0.8到0.9之间。
但新方案又带来了以下新问题：
- 所有模型的创建与部署运行时长超过六个小时。单是这一条，就可能导致在模型准备就绪并部署完成时，其预测已经过时。
- 我们使用Scikit-learn作为训练主框架，但其在设计层面并没有考虑到分布式执行情况。我们无法在Amazon EMR工作节点上充分使用模型训练与测试组件。训练只能在主节点上进行，而在此期间工作节点则完全闲置。我们只能以序列化方式在单一实例之上逐个创建模型。
- 这种方法会带来极为复杂的管道体系，需要配合额外的管理资源才能保持正常运行。在Amazon Elastic Compute Cloud（Amazon EC2）实例上为多模型训练分配资源，还会增加成本与维护压力。
为了解决上述问题，我们开始研究更先进的技术。
Apache Arrow
Spark 2.3以上版本开始集成Apache Arrow，这是一套面向列式内存数据的跨语言开发平台。Apache Spark则是一种JVM语言（用Scala编写），但本次演练中使用的全部代码都只使用Python。
在使用Spark SQL时，两种语言之间的性能没有明显的差异。但如果在JVM与Python解释器之间进行大量序列化，例如编写用户定义函数（UDF）或者在本地收集数据（例如使用toPandas方法）时，就会出现很大问题。
值得注意的是，编写UDF是一种以单行形式进行操作的强大工具，但缺点在于无法通过Spark Catalyst优化器进行优化。
从宏观层面来看，Apache Arrow主要具有以下优势：
内存数据
Apache Arrow是一套面向内存数据的通用型平台。通过图一可以看到，在使用了Arrow内存格式之后，Apache Arrow完全可以充当不同技术（例如Apache Spark、Pandas以及Apache Parquet）之间的连接器。
图一——不同技术间的序列化与反序列化只需要执行一次。
列式内存
作为内存中的列，Apache Arrow能够以远超行式存储的效率实现内存管理，并充分利用现代CPU与GPU的资源优势。图二所示，为RAM中的行式与列式内存（列式内存在分析功能方面具有显著优势）。
图二——CPU中的内存管理。
在PySpark项目当中安装并启用Apache Arrow后，我们将体会到明显的Pyspark DataFrame到Pandas DataFrame转换性能的提升。
启用Apache Arrow之后，系统还能够支持以下两种类型的矢量化udf：
- 标量Pandas UDF可接收pandas.Series形式的输入，并返回pandas.Series形式的输出。
- 分组Map Pandas UDF能够根据操作符所指定的条件将单一Spark DataFrame拆分成多个组，对每个组应用UDF(pandas.DataFrame到pandas.DataFrame)，并将返回的结果组合成新的Spark DataFrame。
我们的解决方案
我们希望根据某一项特征对多个模型同时进行训练与测试，借此优化广告在网络中的实际效果（CTR与VCR）。在我们的工具集中，我们选择Apache Spark on Amazon EMR作为计算引擎。
我们可以将模型的训练拆分为两次Spark提交：
- 为模型训练进行数据准备: 我们可以使用Spark SQL来准备数据，选择特征并制作样本。如前所述，我们希望将每项特征值拟合为对应的单一模型（此特征中包含六个值）。这样，我们就能以优化度更高、更有效的方式处理 模型数量 x 10M的样例数据了。创建的样本将按各个值进行分区，并以Parquet的形式保存在 Amazon Simple Storage Service (Amazon S3)当中。特征的每个样本值都应大小相同，借此防止特征发生偏移。代码示例:
https://gist.github.com/ijan10/d7b11d6906e1b87bedd339b0cc04146f在示例代码中，我们创建了一个样本数据框，并按其中一项特征进行分区。
- 训练分布式模型: 从Amazon S3中读取分区数据，按具体特征进行分组，而后应用于分组Map Pandas UDF。主函数示例代码：https://gist.github.com/ijan10/e14b013819ad2536b30097054dfdcdaa每个Pandas UDF都会从全部特征值样本中获取Pandas DataFrame。Spark分区不会在各执行程序间拆分，且每个Spark分区都将包含用于创建模型的全部样本数据。我们对运行Pandas UDF的各执行程序进行配置，要求其范围内必须包含一个Pandas DataFrame并能够准备特征、拟合模型（Scikit-learn）、为PMML的使用做好准备、进行交叉验证并将输出结果保存至Amazon S3。要了解更多详细信息，请参阅我们发布在GitHub中的 pandas_udf_ctr_create_model.py 。
我们还需要以训练模型的方式完成后期分布式验证。为此，我们在模型部署完成后创建的数据中使用相同的架构（各项模型测试均在Pandas UDF中运行）。
总而言之，我们创建了一套通用性强且简单易行的解决方案，可用于配置并测试每种属性组合的模型，并在不变更代码的前提下并行支持多套模型。
我们的Amazon EMR集群
我们在每个分区内使用最大10M条记录作为批次上限，同时配合具有以下硬件的EMR实例Fleet集群：
- 主实例——4xlarge（r4.4xlarge, i3.4xlarge, r5.4xlarge等）。
- 4个核心实例——4xlarge（r4.4xlarge, i3.4xlarge, r5.4xlarge等）。
对于Pandas UDF，我们希望尽可能增加执行程序数量，并在每个执行程序中都至少使用包含10M条记录（Pandas与Scikit-learn）的批量数据对一套模型进行拟合。要了解更多详细信息，请参阅我们在GitHub上的Spark提交。
4xlarge实例通常配备约122 GB内存与16个vCPU，因此基于上述Spark提交，我们将在每个实例中创建一个执行程序（总计占用110 GB内存与14个vCPU）。
总结
Spark是一套分布式计算框架，通过PyArrow实现了Pandas UDF等多项新功能。我们可以使用Spark的分布式与高级机器学习模型生命周期功能构建起具有大批量生产模型的规模化产品。
在本文中，我们分享了一种实现模型生命周期功能的方法，能够使用少量PySpark代码实现分布式模型训练与测试。通过这项功能，我们也切实提高了机器学习模型的性能与准确度。
本文内容与观点来自第三方作者，AWS对本文的内容或准确性不承担任何责任。
Perion Network – APN合作伙伴聚焦
Perion Network为APN指定技术合作伙伴。其AI团队为客户提供实时性能优化服务、推荐系统与供需预测工具。
*已经与Perion Network开展合作? 评价此合作伙伴
*要对APN合作伙伴做出评价，您必须身为曾直接与其合作完成项目的AWS客户。