出租车数据分析

一、实验简介

图片来自pixabay.com

出租车是我们生活中经常乘坐的一种交通工具，但打车难的问题也限制了我们更好地利用这种交通方式。在哪些地方出租车更容易打到？在什么时候更容易打到出租车？本课程将基于某市的出租车行驶轨迹数据，带你学习如何应用Spark SQL和机器学习相关技巧，并且通过数据可视化手段展现分析结果。

1.1 知识点

• Spark DataFrame操作
• Spark SQL 的 API 查询
• Spark MLlib 的 KMeans 算法应用

1.2 准备工作

本课程需要你具有一定的Spark基础，以下为推荐在本课程之前需要学习的课程（已按先后顺序进行排列）：

• Spark大数据动手实验
• Spark DataFrame入门
• Spark DataFrame详解
• Spark 讲堂之 SQL 入门

二、数据集简介及准备

图片来自pixabay.com

2.1 数据集简介

本数据集为四川省成都市的出租车GPS记录数据集。该数据集已提前清洗完成，仅提取了原始数据集中某一天的部分数据，并且去除了时间段在 0 点至 6 点之间的较少数据。

数据记录了成都市部分出租车在载客时的GPS位置和时间等信息，数据记录的格式为 CSV 格式。

已清洗的数据仅供本课程学习使用，有一定的模拟性质。如需要更多的信息，则需要从原始数据按照相应的目的进行清洗。

该数据集中的一条记录如下所示：

1,30.624806,104.136604,211846

对各个字段逐个解释如下：

• TID：出租车的ID。每辆出租车的TID都是唯一的。
• Lat：出租车状态为载客时的纬度。
• Lon：出租车状态为载客时的经度。
• Time：该条记录的时间戳。如 211846 代表 21 点 18 分 46 秒。

CSV 格式是数据分析工作中常见的一种数据格式。CSV 意为逗号分隔值（Comma-Separated Values），其文件以纯文本形式存储表格数据（数字和文本）。每行只有一条记录，每条记录被逗号分隔符分隔为字段，并且每条记录都有同样的字段序列。

CSV 格式能被大多数应用程序所支持，广泛用于在不同的系统之间转移数据，是一种容易被兼容的格式。实验楼中大量的数据分析类课程都使用了 CSV 格式的数据集，不仅如此，我们也推荐你在今后的数据分析工作中应用此格式来存储数据。

2.2 下载数据集

双击打开桌面上的 Xfce 终端，然后输入下面的命令以下载航班数据集：

wget http://labfile.oss.aliyuncs.com/courses/736/taxi.csv

下载得到的 CSV 数据文件位于你使用解压命令时的工作目录中，默认情况是在 /home/shiyanlou 目录中。

2.3 启动 Spark Shell

为了更好地处理 CSV 格式的数据集，我们可以直接使用由 DataBricks 公司提供的第三方 Spark CSV 解析库来读取。

首先是启动 Spark Shell。在启动的同时，附上参数--packages com.databricks:spark-csv_2.11:1.1.0

请在终端中输入以下代码。

spark-shell --packages com.databricks:spark-csv_2.11:1.1.0

注意：该操作需要联网权限。如果遇到网络访问较慢，或者是您当前不具备访问互联网的权限时，请参考文末的常见问题“无法访问外网时，应如何通过加载 CSV 解析库的方式进入Spark Shell”，问题解答中提供了解决方案。

2.4 导入数据

2.4.1 加载实验所需的包

首先我们需要加载本节实验所需要的包。这些包主要有：

请在 Spark Shell 中输入以下代码。

import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.clustering.KMeans

上述包的使用方法可以查阅 Spark 的 API 手册，地址为：

http://spark.apache.org/docs/latest/api/scala/index.html

将 URL 中的 latest 修改为你当前正在使用的 Spark 版本号（如 1.6.1），则可以查阅与当前版本匹配的 API 手册，从而获得更加精确的说明。

2.4.2 定义字段格式

在实验楼为你提供的大多数的数据分析类课程中，用到的 CSV 格式的数据集都在首行标记了各个字段的名称。但本课程中用到的数据集却没有这个关键信息，如果我们直接创建 DataFrame 的话，就不能够很好地去定位到各个列。因此在导入数据之前，我们需要先定义数据的字段格式（Schema）。

在学习 Spark SQL 时，我们已经知道： Spark SQL 支持两种不同的方式来将现有的 RDD 转换为数据框（DataFrame）。第一个方法是使用反射机制（Relection），另一个则是通过编程的方式指明字段格式。

注：在 2.0 版本及之后的 Spark 中，提出了一个 DataSet 的概念来取代 DataFrame。请关注 Spark 官网提供的 API 文档和编程指南，以适应这个变化，敬请注意。

请在 Spark Shell 中输入以下代码。

// 利用 StructType 定义字段格式，与数据集中各个字段一一映射。
// StructField 中的的三个参数分别为字段名称、字段数据类型和是否不允许为空。
val fieldSchema = StructType(Array(
  StructField("TID", StringType, true),
  StructField("Lat", DoubleType, true),
  StructField("Lon", DoubleType, true),
  StructField("Time", StringType, true)
))

2.4.3 读取数据

定义好字段格式之后，调用了 sqlContext 提供的 read 接口，指定加载格式 format 为第三方库中定义的格式 com.databricks.spark.csv 。因为本次课程使用的数据集中首行没有各列的字段名称，因此需要设置读取选项 header 为 false。最后，在 load 方法中 指明待读取的数据集文件的路径。

注意，因为我们是在 Spark Shell 中以交互式命令行的形式输入代码，Spark Shell 在启动过程中就已经创建好了 sqlContext 对象，我们可以直接使用。如果你是在以独立应用的方式开发 Spark 程序，请手动通过 Spark Context 来创建 SQL Context 。

请在 Spark Shell 中输入以下代码。

val taxiDF = sqlContext.read.format("com.databricks.spark.csv").option("header", "false").schema(fieldSchema).load("/home/shiyanlou/taxi.csv")

可以看到读取后的数据已经映射到各个定义好的字段中了。

2.4.4 检查已导入的数据

读取数据之后，通常会使用 printSchema() 方法打印出 DataFrame 的字段格式。

请在 Spark Shell 中输入以下代码。

taxiDF.printSchema()

使用 show() 方法打印出前 20 条记录，查看数据是否正常。

请在 Spark Shell 中输入以下代码。

taxiDF.show()

三、对出租车数据进行聚类

图片来自pixabay.com

3.1 K-Means 聚类算法简介

我们常说“物以类聚，人以群分”，将物理或抽象对象的集合分成由类似的对象组成的多个类的过程，则被称为聚类。

聚类的用途有很多，比如：

• 在商务上，聚类能帮助市场分析人员从客户基本库中发现不同的客户群，并且用购买模式来刻画不同的客户群的特征。
• 在生物学上，聚类能用于推导植物和动物的分类，对基因进行分类，获得对种群中固有结构的认识。
• 在房屋租售方面，可以根据房子的类型、价值和地理位置对一个城市中房屋的分组，从而更好地进行租售信息和价格水平的维护。
• 在保险业，可以通过识别可能的欺诈行为，找出一段时间内索赔支出很高的投保人，通过拒保等措施保护保险公司的正当权益。

而 K-Means 算法是一个迭代型的聚类算法。迭代是指需要一个或者多个往复的阶段才能完成，结束的条件是找到最优的簇。

算法的过程可以描述如下：

1. 给定簇的数量 K 和一个数据集合（包含 N 个点，点可以是多维的）。在数据集合中，随机产生 K 个初始化的均值（本例中 K = 3）作为质心。下图中被标记为彩色的点即为初始质心。 
2. 计算集合内各个点与这 K 个中心点的距离，并且将各个点分配到与它距离最近的质心。 
3. 每个 K 集群的质心成为新的均值。 
4. 当所有的点都被分配之后，再重新计算 K 个质心的位置，然后重复上述 2 、3 步骤，直到质心不再改变时，该算法结束。 

上述步骤中的图片均来自维基百科。

3.2 Spark 框架中的 K-Means 算法实现

Spark 框架在 MLlib 库中为广大的使用者提供了一些常用的机器学习算法的实现。MLlib中包含的机器学习算法主要有：

• 回归
• 二元分类
• 聚类
• 协同过滤
• 梯度下降优化

K-Means 就是聚类算法中的一种，它可以将数据点进行分组。在 Spark MLlib 中， 实现了 K-Means 算法的并行化版本。因此我们在 Spark 中进行此类问题的分析，能够获得相比于 R 语言、Matlab 等分析工具的更好的运算速度。

在 Spark 的 Github 页面 可以看到 K-Means 算法在 Spark 中的实现细节，如果你有兴趣的话不妨花点时间去阅读它的代码。代码中给出了较为详细的注释，阅读算法的实现细节能够帮助你更加深刻地理解其原理和各项参数的意义。

3.3 定义特征数组

为了能够使用 Spark 框架实现的 K-Means 算法，我们需要将数据的特征按照这类机器学习算法的要求进行转换。

在机器学习算法应用过程中，经常面临的工作便是为机器学习算法提供的 API 的各项参数 “制备” 数据。

转换的目标格式是特征向量，它用数字代表每种类型的特征值（例如用 1 代表位置 A，用 2 代表位置 B ）。这样做的好处是为了简化数据本身不同特征类型（字符型、数值型、日期时间型等）之间带来的复杂度提升。我们可以在 Spark 中利用 VectorAssembler 这个 API 来进行转换。 它会返回在单个列向量中包含所有特征列的 DataFrame 。

对于出租车数据，我们在聚类这一步中主要对其位置进行分析，因此只需要将纬度和经度这两个特征转换为一个特征数组即可。

请在 Spark Shell 中输入以下代码。

val columns = Array("Lat", "Lon")

然后创建一个向量装配器 VectorAssembler 对象，并设置相关的属性。并利用向量装配器对象的 transform() 方法对导入的数据（taxiData）进行转化。VectorAssembler 是一个能够将多个列合并到单个列向量中的特征转化器。

可以查阅 API 手册来了解如何使用 VectorAssembler 。

请在 Spark Shell 中输入以下代码。

// 设置参数
val va = new VectorAssembler().setInputCols(columns).setOutputCol("features")

// 将数据集按照指定的特征向量进行转化
val taxiDF2 = va.transform(taxiDF)

按照惯例，仍然需要对转化后的 DataFrame 中的数据进行检查。

请在 Spark Shell 中输入以下代码。

taxiDF2.show()

3.4 进行 K-Means 计算

在 Spark 的 API 手册中，如果你翻到 KMeans 部分，会发现这样的一句话：

由于 K-Means 算法是迭代算法，需要对数据多次操作，因此 Spark 官方文档中建议我们将其缓存下来以加速计算。

请在 Spark Shell 中输入以下代码。

taxiDF2.cache()

聚类之前，我们先把数据集按照合适的比例划分为训练集和测试集。此处我们设置 DataFrame 中 70% 的数据为训练集， 30% 的数据为测试集。同时，使用 randomSplit() 方法对其进行划分。

请在 Spark Shell 中输入以下代码。

// 设置训练集与测试集的比例
val trainTestRatio = Array(0.7, 0.3)

// 对数据集进行随机划分，randomSplit 的第二个参数为随机数的种子
val Array(trainingData, testData) = taxiDF2.randomSplit(trainTestRatio, 2333)

数据准备好之后，就可以创建 KMeans 对象并对指定的数据进行训练。

涉及到的 API 主要有：

• setK()：是一个 “Parameter setter”，用于设置聚类的簇数量。
• setFeaturesCol()：设置数据集中的特征列所在的字段名称。
• setPredictionCol：设置生成预测值时使用的字段名称。
• fit()：将 KMeans 对象对指定数据的特征进行匹配适应，训练模型。

请在 Spark Shell 中输入以下代码。注意：此步骤比较耗时，请耐心等待。

// 设置模型的参数
val km = new KMeans().setK(10).setFeaturesCol("features").setPredictionCol("prediction")

// 训练 KMeans 模型，此步骤比较耗时
val kmModel = km.fit(taxiDF2)

获取 KMeans 模型的聚类中心。

请在 Spark Shell 中输入以下代码。

val kmResult = kmModel.clusterCenters

可以看到输出的结果中，便是我们设定数量为 10 的聚类结果。

请将当前这一步得到的结果保存到文件中，我们在下一节的数据可视化实验中会用到这些数据。

请在 Spark Shell 中输入以下代码。

// 先将结果转化为 RDD 以便于保存
val kmRDD1 = sc.parallelize(kmResult)

// 保存前将经纬度进行位置上的交换
val kmRDD2 = kmRDD1.map(x => (x(1), x(0)))

// 调用 saveAsTextFile 方法保存到文件中，
kmRDD2.saveAsTextFile("/home/shiyanlou/kmResult")

3.5 对测试集进行聚类

接下来，我们使用已经训练好的 KMeans 模型，对测试集中的数据进行聚类，产生预测的结果，并对聚类结果进行深入分析。

首先是调用模型的 transform() 方法对测试数据进行聚类。

请在 Spark Shell 中输入以下代码。

val predictions = kmModel.transform(testData)

然后查看预测结果。

请在 Spark Shell 中输入以下代码。

predictions.show()

3.6 分析聚类的预测结果

预测结果的类型为 DataFrame ，我们先将其注册为临时表以便于使用 SQL 查询功能。

请在 Spark Shell 中输入以下代码。

predictions.registerTempTable("predictions")

在先前的课程中，我们都是通过编写 SQL 语句来进行查询。这一次，尝试使用 SQL 相关的 API 来完成同样的工作。

3.6.1 每天哪个时段的出租车最繁忙？

为了回答这个问题，需要在预测结果中提取出 Time 字段的前 2 位作为一天之中的小时数，同时提取 prediction 字段为后续的步骤做准备。

请在 Spark Shell 中输入以下代码。

/* 使用 select 方法选取字段，
*  substring 用于提取时间的前 2 位作为小时，
*  alias 方法是为选取的字段命名一个别名，
*  选择字段时用符号 $ ，
*  groupBy 方法对结果进行分组。
*/
val tmpQuery = predictions.select(substring($"Time",0,2).alias("hour"), $"prediction").groupBy("hour", "prediction")

接着，我们基于上述查询的结果，对每个小时不同预测类型的数量进行统计。

请在 Spark Shell 中输入以下代码。

/* agg 是聚集函数，count 为其中的一种实现，
*  用于统计某个字段的数量。
*  最后的结果按照预测命中数来降序排列（Desc）。
*/
val predictCount = tmpQuery.agg(count("prediction").alias("count")).orderBy(desc("count"))

最后，输入前 20 条记录，以查看预测的结果。

请在 Spark Shell 中输入以下代码。注意：此步骤比较耗时，请耐心等待。

predictCount.show()

结果中每个时段的出租车服务次数按照降序进行了排列。可以看到 14点 至 17 点这个时段里，聚类区域为 9 号区域内的出租车载客次数是最多的，为 6018 次。同时，从总体上来看，聚类区域为 9 号的区域前 20 条记录内占绝大多数，我们可以大胆推测在该区域内打车的人比较多，并且在 14 点、 15 点 和 17 点 最难打到车（真的是这样吗？根据自己的生活经验想一下是否合理）。

为了弄清楚 9 号区域具体是什么地方，可以先将上述结果保存下来，我们在下一节的数据可视化实验中来揭晓答案。

请在 Spark Shell 中输入以下代码。注意：此步骤比较耗时，请耐心等待。

predictCount.save("/home/shiyanlou/predictCount", "com.databricks.spark.csv")

3.6.2 每天哪个区域的出租车最繁忙？

其实我们在回答前一个问题的时候，这个问题的答案已经隐藏在其中了，我们只需要一个查询语句便可得到。

请在 Spark Shell 中输入以下代码。

val busyZones = predictions.groupBy("prediction").count()

busyZones.show()

同样，将查询结果予以保存，在下一节数据可视化过程中为其绘制图表。

请在 Spark Shell 中输入以下代码。注意：此步骤比较耗时，请耐心等待。

busyZones.save("/home/shiyanlou/busyZones", "com.databricks.spark.csv")

四、实验总结

在本课程中，我们通过 Spark 对成都市出租车的行驶轨迹数据进行分析，找出了高峰时段和繁忙的服务区域。

在下一节的实验中，我们将继续利用本节实验得到的数据进行数据可视化的学习。

如果你在学习过程有任何疑问或者建议，都欢迎到实验楼的讨论区与我们交流。

作业

在本节实验中，我们尝试使用 SQL 的 API 方式进行了数据的查询。那么你能否将 3.6.1 和 3.6.2 小节中涉及到的查询语句用 SQL 语句表达出来呢？

例如：

sqlContext.sql("SELECT * FROM predictions")

常见问题

• Q：无法访问外网时，应如何通过加载CSV解析库的方式进入Spark Shell？
• A：请按照下面的步骤进行配置。
  1. 在终端中输入命令： wget http://labfile.oss.aliyuncs.com/courses/610/spark_csv.tar.gz 下载相关的 jar 包。
  2. 将该压缩文件解压至 /home/shiyanlou/.ivy2/jars/ 目录中，确保该目录含有如图所示的以下三个 jar 包。 
  3. 在终端中输入命令 spark-shell --packages com.databricks:spark-csv_2.11:1.1.0 启动即可。