Python技术交流与分享

2.1机器学习系统架构

Spark发展非常快，到我们着手编写本书时，Spark已升级为2.1版，这2.0以后，Spark大大增强了数据流水线的内容，数据流水线的思路与SKLearn非常相似，我想这种思路或许是未来的一个趋势，使机器学习的流程标准化、规范化、流程化，很多原来需要自己编写代码都有现成的模块或函数，模型评估、调优这些任务也可实现了更高的封装，这大大降低机器学习门槛。

图2-1 Spark机器学习系统的架构图
其中数据处理、建模训练，我们可以进行组装成流水线方式，对模型评估及优化可以采用自动化方式。

2.2启动集群

Spark集群的安装配置，这里不做详细介绍，我们提供了本书可操作云平台，对Spark集群的安装配置感兴趣的读者，可参考由我们编写的《自己动手做大数据系统》。
Spark运行方式有本地模式、集群模式，本地模式所有的处理都运行在同一个JVM中，而后者，可以运行在不同节点上。具体运行方式主要有：
表2-1 Spark运行模式

本书主要以Spark Standalone(独立模式)为例，如果想以其他模式运行，只要改动对应参数即可。
Spark支持Scala或Python的REPL(Read-Eval-Print-Loop,即交互式shell)来进行交互式程序编写，交互式编程，输入的代码执行后立即能看到结果，非常友好和方便。
在2.0之前的Spark版本中，Spark shell会自动创建一个SparkContext对象sc。SparkContext与驱动程序(Driver Program)和集群管理器（Cluster Manager）间的关系如图2-2所示：

图2-2 SparkContext与驱动程序、集群管理器间的关系图

从图中可以看到SparkContext起中介的作用，通过它来使用Spark其他的功能。每一个JVM都有一个对应的SparkContext，Driver program通过SparkContext连接到集群管理器来实现对集群中任务的控制。Spark配置参数的设置以及对SQLContext、HiveContext和StreamingContext的控制也要通过SparkContext。
不过在Spark 2.0中引入SparkSession对象（spark），运行Spark shell则自动创建一个SparkSession对象，在输入spark时就会发现它已经存在了（参考图2.图2-3），SparkConf、SparkContext和SQLContext都已经被封装在SparkSession当中，它为用户提供了一个统一的切入点，同时也提供了各种DataFrame和Dataset的API，大大降低了学习Spark的难度。

图2-3 启动Spark shell界面

图2-3是启动Spark的集群的界面，编程语言是Scala，如果希望使用Python为编辑语句，该如何启动呢？运行pyspark即可。

图2-4 启动PySpark的客户端

2.3加载数据

这里以MovieLens 100k（http://files.grouplens.org/datasets/movielens/ml-100k.zip）数据集中的用户数据（u.data）为例，首先在本地查看数据的基本信息，然后把本地文件复制到HDFS上，Spark或PySpark读取读取hdfs上的数据。
查看u.user文件的基本信息，数据样例，总记录数等信息。

u.user用户数据每列的含义为：user id | age | gender | occupation | zip code，即用户ID，用户年龄，用户性别、用户职位、所在地邮编等信息，列间的分隔符为竖线（|）,共有943条记录。
如何把用户信息复制到HDFS上？首先，查看当前HDFS的目录信息。

由此可知在HDFS已有/u01/bigdata/data目录（如果没有目录可以通过hadoop fs -mkdire /u01/bigdata/data命令创建。），通过以下命令，把本地文件u.user复制到HDFS上。

把电影评级数据（u.data）、电影数据(u.item)等复制到HDFS方法相同，把本地数据复制到HDFS后，Spark如何读取加载HDFS上的文件？我们可以通过Spark的textFile方法读取。这里我们以PySpark为例，启动PySpark客户端，导入需要是的包，然后通过textFile方法读取HDFS上的数据，具体请看以下示例：

2.4探索数据

生产环境中数据往往包含很多脏数据，如缺失数据、不一致、不规范、奇异数据等等，所以数据加载后，数据建模前，需要对数据进行分析或探索，尤其面对大数据，了解数据的统计信息、数据质量、数据特征等，为数据处理、数据建模提供重要依据，在进行这些数据分析时，如果能实现数据的可视化，当然更利于我们理解数据。
2.4.1数据统计信息
加载数据后，首先关注的数据的统计信息，有了数据统计信息，我们对数据就有了一个大致了解，如数据特征的最大值、最小值、平均值、分位数、方差等。这些信息有助于我们理解数据质量、数据构成，为数据预处理提供重要依据。
#查看用户各字段的统计信息

从以上统计可以看出，用户表总记录数为943条，年龄最小为9岁，最大为73岁，平均年龄为34岁。
2.4.2数据质量分析
数据质量分析是数据探索阶段重要一环，数据不是完美的，大多数据大多包含缺少数据、不一致数据、异常数据、噪音数据等。没有可信的数据，再好的模型性能都太可能好，正所谓“垃圾进，垃圾出”。
数据质量方面的分析，主要包括以下几个方面：
1）缺失值;
2）异常值；
3）不一致的值
4）错误数据
数据集下载
本节以一份某酒店的销售额的数据为例，来说明在数据探索中，对数据质量的一般分析方法，主要涉及缺少值、异常值、不一致数据等。

图2-5 销售额箱型图检测异常值

从以上分析，可知，销售额列存在两个空值、6个可能的异常值，其中865.0,1060.0有可能属于正常值，当然也需要和也相关业务员沟通，对其他异常值，需要进一步分析异常值产生的原因，然后，确定数据的去留。
2.4.3数据特征分析
对数据质量有基本了解后，接下来就可就数据的特征进行分析，数据特征分析一般包括以下一些内容：
特征分布分析
对比分析
统计量分析
特征一般指用于模型训练的变量，原始数据中特征，有些是数值，有些是字符或其他格式信息，但在进行机器学习前，都需要转换为数值。根据实际情况，有时需要根据已有特征生成或衍生出新特征，如根据用户年龄衍生出表示老、中、青的新特征；有时需要对一些特征进行规范化、标准化等转换，尤其对回归类模型。
2.4.3.11.数据特征分析
特征的分布分析有助于发现相关数据的分布特征、分布类型、分布是否对称等，可以使用数据可视化方法，易直观发现特征的异常值等。以用户信息数据为例，分析用户的年龄特征、职业特征等。

图2-6 用户年龄特征分布图

从以上图形可以看出，最小年龄在10岁左右，最大年龄超过70岁，大部分是20岁到40岁之间。
我们还可以进一步分析用户职业分布特征。

图2-7 用户职业分布图

从以上用户职业分布图，可以看出，学生占绝大多数，其次其他职业、教育工作者、管理者、工程师等。医生、家庭主妇或许平时较忙，故数量比较少。
2.4.3.22.特征分布及相关性分析
在数据探索阶段，分析特征分布，特征间的相关性等，对应后续的特征选择、特征提取将提供重要依据，以下是对类似共享单车数据的特征分析，详细内容可参考第9章的9.3节

图2-8 hours数据集特征分布及相关性示例图
2.4.3.33.对比分析

图2-9销售月份对比图
2.4.4数据的可视化
数据的可视化是数据探索、数据分析中重要任务，通过可视化可帮助我们发现数据的异常值、特征的分布情况等，为数据预处理提供重要支持。Spark目前对数据的可视化功能还很弱或还没有，不过，没关系，我们可以借助Python或R等可视化功能，Python和R在数据可视化方面功能很强大，这里以Python的数据可视化为例。Python的数据表现能力很强，可以2D或3D等方式展示，视化可以使用matplotlib或plot等方法。matplotlib是一种比较低级但强大的绘图工具，可以进行很多定制化，但往往需要较大代码来实现;Plot是一种非常简洁的绘图工具，它主要基于pandas基础之上，以下我们通过两个示例来具体说明：
下例是通过matplotlib可视化sin(x)和cos(x)函数的图形。

运行结果如下：

图2-10 matplot数据可视化
同样的这些数据，如果我们对这些数据使用plot来进行可视化，代码可以非常简洁，但定制化方面可能要弱一些。

显示图形如图2-11所示下：

图2-11 plot数据可视化

从以上实现代码可以看出，如果使用plot则非常简单，虽然定制化要比matplotlib少些，但其可定制的项也不少，如kind，rot，title，legend等等。

2.5数据预处理

前面我们介绍了探索数据的一些方法，通过对数据的探索，可以帮助我们发现一些奇异值、缺失值、一些特征的类别及其分布情况等信息。而这些信息正是对数据预处理的重要依据。在数据分析、机器学习中，数据的预处理是一个非常关键、尤其是涉及大数据的处理，往往是比较费时、费神的一个过程，有时，还需要往返多次。当然，如果数据预处理得好，除提高数据质量外，更能极大提高模型的性能，反之，对模型的影响也是很大，甚至可能垃圾进，垃圾出。
数据的预处理一般包括数据清理、数据转换、数据集成、数据归约等。这些预处理主要内容可以通过以下图形2-12来表示：

图2-12数据预处理示意图
2.5.1数据清理
数据清理主要任务是填补缺失值、光滑噪声数据、处理奇异数据、纠正错误数据、删除重复数据、删除唯一性属性、去除不相关字段或特征、处理不一致数据等。噪声数据的处理方法：分箱、聚类等。以下分别以处理缺失数据、异常数据为例，说明在spark中如何处理。
1. 处理缺失值

2. 处理奇异值
在数据探索阶段，我们发现销售数据文件catering_sale.csv中有6个可能的奇异值，假设与相关人员核实后，只有22为奇异值或错误数据，对错误数据我们一般采用删除或替换的方法，这里我们采用Spark SQL来处理奇异数据。
首先把数据复制到HDFS，用Spark读取数据，如果启动pyspark，则可以通过spark.read.csv("/home/hadoop/data/catering_sale.csv",header=True)读取；如果启动spark-shell启动，则可以采用 spark.read.option("header","true").csv("hdfs://192.168.1.112:9000/home
/hadoop/data/catering_sale.csv")的方式读取。

这里我们使用了DataFrame的select、replace等方法，实际上df还有很多可利用的方法或函数，可以通过df.+Tab键查看：

这些方法或函数的具体使用，可以通过df.方法名?的方式查看，下例为查看df.filter的详细用法：

此外，我们还可以使用大量spark.sql.functions或pyspark.sql.functions,以下是使用去除字段左右空格、截取字段长度等内置函数示例：

2.5.2数据变换
数据变换是数据预处理中一项重要内容，如对数据进行数据的规范化、离散化、衍生指标、类别特征数值化、平滑数据等都属于数据变换。数据变换Spark ML有很多现成的算法，利用这些算法可极大提高整个数据处理的效率，下表2-2只是为一个概况，更多更详细信息请可参考第4章。
表2-2 Spark ML自带的数据变换算法

这里我们以卡方检验为例，如何根据特征的贡献率来选择特征。假设我们很多特征，如：表示时间的特征：季节（season）、年月（yr）、月份（mnth）、是否节假日（holiday）、是否周末（weekday）；表示天气的特征weathersit，temp等等，为了使用卡方检验来选择这些特征，首先需要把各特征组合一个特征向量，然后，把整合后特征向量、及选择特征个数等代入卡方模型中，详细代码如下：

2.5.3数据集成
数据集成是数据预处理的重要内容之一，将多文件或者多数据库中的数据进行合并，然后存放在一个一致的数据存储中。数据集成一般通过join或union、merge等关键字把两个（或多个）数据集连接在一起，Spark SQL(包括DataFrame)有join方法，Pandas下有merge方法。数据集成往往需要耗费很多资源，尤其是大数据间的集成涉及到shuffle过程，有时需要牵涉到多个节点，数据集成除了数据一致性外，性能问题常常不请自来，需要我们特别留心。
传统数据库一般是单机上采用hash join方法，如果在分布式环境中，采用join时，可以考虑充分利用分布式资源进行平行化，当然，在进行join之前，对数据过滤或归约也是常用的优化方法。
Spark SQL中有三种join方法：
broadcast hash join：
如果join的表中有一张大表和一张较少的表，可以考虑把这张小表广播分发到另一张 大表所在的分区节点上，分别并发地与其上的分区记录进行hash join。
shuffle hash join：
如果两张表都不小，对数据量较大的表进行广播分发就不太适合。这种情况下，可以根 据join key相同必然分区相同的原理，将两张表分别按照join key进行重新组织分区， 这样就可以将join分而治之，划分为很多小join，充分利用集群资源并行化。
sort merge join：
如果两张表都比较大，可以考虑使用sort merge join方法，先将两张大表根据join key进行重新分区，两张表数据会分布到整个集群，以便分布式并行处理，然后，对单个分区节点的两表数据，分别进行排序，最后，对排好序的两张分区表数据执行join操作。
当然，如果两表都不大，可以直接使用hash join。
DataFrame中join有（或merge）：内连接、左连接、右连接等。
2.5.4数据归约
大数据是机器学习的基础，但大数据往往数据量非常大，有时我们可以通过数据归约技术，删除或减少冗余属性（或维）、精简数据集等，使归约后数据比原数据小或小得多，但仍然接近于保持原数据的完整性，并结果与归约前结果相同或几乎相同。
表2-3 Spark ML 自带的数据选择算法

选择特征或降维是机器学习中重要的处理方法，我们可以使用这些方法在减少特征个数、消除噪声等问题的同时，维持原始数据的内在结构或主要特征。尤其是降维，在大数据、机器学习中发挥中重要作用，以下通过两个实例说明SVD、PCA具体使用。目前Spark MLlib支持SVD及PCA。

同样这个矩阵data，以下我们用PCA进行分解，看一下效果及与SVD的异同，SVD分解后右奇异矩阵V与PCA降维后的矩阵pc很相似。

使用PCA降维，利用pyspark的画图功能，可以新生成的特征的方差贡献度进行可视化，下图为对hour.csv数据，通过PCA处理后，重要特征的排序情况：

图2-13 hour.csv数据的PCA分析图

2.6构建模型

前面我们介绍了准备阶段，包括加载数据、探索数据、预处理数据等，数据准备阶段往往是最费时间和精力的，常常这个问题解决了，又会出现新问题，经常需要返回多次。一般而言数据准备阶段从时间上来说可能要占据60%左右，有时更多。 数据准备后以后，接下来就是构建模型，模型是机器学习、数据挖掘等的核心，构建模型涉及确定模型或算法、设置参数、运算模型等，其大致流程如图2.11-14所示。

图2.14构建模型流程

选择算法主要依据业务需求、数据特征等，Spark目前支持分类、回归、推荐等这些常用而且重要的算法，具体可参考表2.3 所示。一种类型往往有几种算法，如分类可以逻辑、决策树等，如何选择算法，需要考虑业务需求、数据特征、算法适应性、个人经验等，当然，也可选择几种方法，然后进行比较，或采用集成学习的方式，复合多种算法也是选项之一，如先采用聚类方法对数据进行聚类，然后对不同类别的数据进行预测或推荐，有时会得到更好的结果。如果你觉得选择比较难或还不好确定，可以先从简单或熟悉的方法开始，然后，不断完善和优化。

表2-3 Spark ML 目前支持的算法

确定算法后，一般还需要设置一些参数，如训练决策树时需要选择迭代次数、纯度计算方法、树的最大高度等，此外，对准备好的数据需要进行划分，一般划分为训练数据和测试数据，有的会把训练数据进一步划分为训练数据集、验证数据集。Spark 提供多种随机划分数据的方法，如randomSplit、CrossValidator等。这些方法的具体使用在2.8节模型调优中将会具体说明。 训练数据用于训练模型，测试数据用于验证模型，因这个环节的验证是在模型训练过程中验证，所以它一般也认为隶属于模型建立过程。这种验证方法一般称为交叉验证（CrossValidator，CV）,有些交叉验证把数据分成K组，如K折交叉验证（K-fold Cross Validator，K-CV ）,在K折-交叉验证中，采用不重复地随机将数据集划分为K对，如果K=3，则将产生3个（训练，测试）数据集对，每个数据集使用2/3的数据进行训练，1/3进行测试。,这样会得到3个模型,用这3个模型的平均数作为最终模型的性能指标。K-CV可以有效的避免欠学习状态的发生,其结果也比较具有说服性。

2.7模型评估

模型构建以后，接下来就需要对该模型的性能、与目标的切合度等进行一些评估，模型评估是模型开发过程的不可或缺的一部分。在构建模型的过程中，会产生一些评估指标，如精确度、ROC、RMSE等等，这这些指标是重要而且基础的，但应该不是唯一和最终指标，除了这些指标外，我们还应该评估模型对业务的提示或商业目标的达成等方面贡献。一个好的模型不但要有好的技术指标，更要为解决实际问题提供帮助，有时后者显得更为重要。 Spark中常用的几个评估算法有： 均方差（MSE,Mean Squared Error）： (∑(prec-act)**2)/n(prec为预测值，act为实际值，n为总样本数) 均方根差（RMSE,Root Mean Squared Error）: 就是MSE开根号 平均绝对值误差(MAE,Mean Absolute Error): (∑|prec-act|)/n 在了解正确率、准确率之前，我们先看一个所谓的混淆矩阵(confusion matrix):

2.15混淆矩阵

混淆矩阵是一个简单矩阵，用于展示一个二分类器的预测结果，其中，T-True、F-False、N-Negative、P-Postitive。 真正(TP）被模型预测为正的正样本数；可以称作判断为真的正确率; 真负(TN) 被模型预测为负的负样本数；可以称作判断为假的正确率; 假正(FP) 被模型预测为正的负样本数；可以称作误报率; 假负(FN) 被模型预测为负的正样本数；可以称作漏报率. 正确率（Accuracy）： A = (TP + TN)/(P+N) = (TP + TN)/(TP + FN + FP + TN) 反映了分类器统对整个样本的判定能力——能将正的判定为正，负的判定为负。 错误率（Error）： E= (FP + FN)/(P+N) = (FP + FN)/(TP + FN + FP + TN) 准确率（Precision） P = TP/(TP+FP) ; 反映了被分类器判定的正例中真正的正例样本的比重 召回率(Recall): R = TP/(TP+FN) = 1 - FN/T; 反映了被正确判定的正例占总的正例的比重 F1-Measure: F1=2P*R/(P+R) 真阳性率(TPR): TPR= TP/(TP+FN),代表分类器预测的正类中实际正实例占所有正实例的比例。 假阳性率(FPR): FPR= FP/(FP+TN)，代表分类器预测的正类中实际负实例占所有负实例的比例。 以上这些都属于静态的指标，当正负样本不平衡时它会存在着严重的问题。极端情况下比如正负样本比例为1:99（有些领域并不少见），那么一个分类器只要把所有样本都判为负，它就拥有了99%的精确度，但这时的评价指标是不具有参考价值的。另外，很多分类器都不是简单地给出一个正或负（0或1）的分类判定，而是给出一个分类的倾向程度，比如贝叶斯分类器输出的分类概率。对于这些分类器，当你取不同阈值，就可以得到不同的分类结果及分类器评价指标，依此人们又发明出来ROC曲线以及AUC（ROC曲线包围面积）指标来衡量分类器的总体可信度。ROC曲线将FPR和TPR定义为x和y轴，这样就描述了真阳性和假阳性不同决策阈值下之间的关系。AUC越大说明模型性能越好，ROC曲线如下图：

图2-16 ROC曲线示意图 下面通过一个实例说明Spark一些评估指标的使用：

准确率：TP/(TP+FP)=17/(17+0)=1
召回率：TP/(TP+FN) = 17/(17+1)=0.944444

2.8组装

我们对数据集进行了探索，之后进行大量的数据清理、转换等工作，对数据预处理后，构建模型、评估模型。评估模型前我们需要对数据集随机划分为训练集和测试集。假如数据有变化，如新增数据，如何保证训练集和测试集上的操作保持一致？如果数据清理、数据转换等有很多步骤，如何保证这些步骤依次执行？
采用Spark pipeline能很好解决这些问题。我们只要把这些任务，作为pipeline的stage，按照其本身的执行次序把这些stages组装到一个pipeline上。（当然如果任务比较复杂，我们也可以采用多个pipeline，然后把这些作为pipeline的stage，组装到一个新的pipeline。）
组装的步骤一般是：
1、创建pipeline，并各个stages依次组装在一起，如：

2、在训练集上拟合这个pipeline

3、在测试集上，做预测。

通过这种方式，既可保证stages有序执行，也可保证在训练集和测试集上所做逻辑操作的一致性，这里只是举了一个简单例子，下一章将详细介绍有关Pipeline的内容，第7章后，还有详细的使用实例。

2.9模型选择或调优

在ML中一个重要的任务就是模型选择，或者使用给定的数据为给定的任务寻找最适合的模型或参数。这个过程也叫做调优。调优可以是对单个的Estimator，比如LogisticRegression，或者是包含多个算法、特征工程和其他步骤的工作流（Pipeline）中完成。用户可以一次性对整个Pipline进行调优，而不必对Pipline中的每一个元素进行单独的调优。
MLlib支持使用像CrossValidator和TrainValidationSplit这样的工具进行模型选择。这些工具需要以下的组件：
Estimator：用户调优的算法或Pipline。
ParamMap集合：提供参数选择，有时也叫作用户查找的参数网格(parameter grid),参数网格可以使用ParamGridBuilder来构建。
Evaluator：衡量模型在测试数据上的拟合程度。
模型选择工具工作原理如下：
1.将输入数据划分为训练数据和测试数据。
2. 对于每个（训练，测试）对，遍历一组ParamMaps。用每一个ParamMap参数来拟合估计器，得到训练后的模型，再使用评估器来评估模型表现。
3.选择性能表现最优模型对应参数表。
2.9.1 交叉验证 （CrossValidator）
交叉验证（CrossValidator）会从将数据集切分成K折数据集合，分别用于训练和测试，。例如，K=3折时，CrossValidator会生成3个（训练数据，测试数据）对，每一个数据对的训练数据占2/3，测试数据占1/3。为了评估一个ParamMap，CrossValidator 会计算这三个不同的（训练，测试）数据集对在Estimator拟合出的模型上的平均评估指标。
在找出最好的ParamMap后，CrossValidator 会利用此ParamMap在整个训练集上可以训练（fit）出一个泛化能力强，误差相对小的的最佳模型，整个过程处于流程化管理之中，其工作流程图如下：

图2-17Spark CrossValidator流程图

虽然利用CrossValidator来训练模型，可以提升泛化能力，但其的代价也比较高，如选择k=3,regParam=(0.1,0.01)，numIters=(10,20)这样就需要对模型训练3*2*2=12次。然而，对比启发式的手动调优，这是选择参数的行之有效的方法。
2.9.2训练-验证切分（TrainValidationSplit）
交叉验证的代价比较高昂，为此Spark也为超参数调优提供了训练-验证切分（TrainValidationSplit）。TrainValidationSplit创建单一的（训练，测试）数据集对。它使用trainRatio参数将数据集切分成两部分。例如，当设置trainRatio=0.8时，TrainValidationSplit将会将数据切分80%作为数据集，20%作为验证集，来生成训练、测试集对，并最终使用最好的ParamMap和完整的数据集来拟合评估器。
相对于CrossValidator对每一个参数进行k次评估，TrainValidationSplit只对每个参数组合评估1次。因此它的评估代价没有这那么高，但是当训练数据集不够大的时候其结果相对不够可信。

图2-18 Spark TrainValidationSplit流程图

2.10保存模型

训练、优化模型后，我们需要保存模型，然后把模型移植或部署到其他环境中。
这节主要介绍如何保存模型，如何部署模型等内容，以下是具体示例代码。
1)保存拟合后流水线（pipeline）到磁盘

2)保存未拟合的流水线（pipeline）到磁盘

3)把拟合后流水线部署到其他环境中。

2.11小结

这一章主要介绍了如何构建Spark学习系统、构建的一般步骤等。，实际上，构建Spark学习系统与我们构建其他平台的学习系统基本相同或相似，一般都包括数据加载、数据探索、数据预测、建模、训练模型、评估模型、优化模型等步骤，但这里我们特别增加一个利用pipeline组装各个任务（stages），这也是Spark ML中基于DataFrame数据集的重要内容，下一章我们将详细介绍有关pipeline的内容。