Uber永久定位系统实时数据分析过程实践！

根据Gartner所言，到2020年，每个智慧城市将使用约13.9亿辆联网汽车，这些汽车配备物联网传感器和其他设备。城市中的车辆定位和行为模式分析将有助于优化流量，更好的规划决策和进行更智能的广告投放。例如，对GPS汽车数据分析可以允许城市基于实时交通信息来优化交通流量。电信公司正在使用移动电话定位数据，识别和预测城市人口的位置活动趋势和生存区域。

本文，我们将讨论在数据处理管道中使用Spark Structured Streaming对Uber事件数据进行聚类分析，以检测和可视化用户位置实践。(注：本文所用数据并非Uber内部实际用户数据，文末附具体代码或者示例获取渠道)

首先，我们回顾几个结构化流媒体涉及的概念，然后探讨端到端用例：

使用MapR-ES发布/订阅事件流

MapR-ES是一个分布式发布/订阅事件流系统，让生产者和消费者能够通过Apache Kafka API以并行和容错方式实时交换事件。

流表示从生产者到消费者的连续事件序列，其中事件被定义为键值对。

topic是一个逻辑事件流，将事件按类别区分，并将生产者与消费者分离。topic按吞吐量和可伸缩性进行分区，MapR-ES可以扩展到非常高的吞吐量级别，使用普通硬件可以轻松实现每秒传输数百万条消息。

你可以将分区视为事件日志：将新事件附加到末尾，并为其分配一个称为偏移的顺序ID号。

与队列一样，事件按接收顺序传递。

但是，与队列不同，消息在读取时不会被删除，它们保留在其他消费者可用分区。消息一旦发布，就不可变且永久保留。

读取消息时不删除消息保证了大规模读取时的高性能，满足不同消费者针对不同目的(例如具有多语言持久性的多个视图)处理相同消息的需求。

Spark数据集，DataFrame，SQL

Spark数据集是分布在集群多个节点上类对象的分布式集合，可以使用map，flatMap，filter或Spark SQL来操纵数据集。DataFrame是Row对象的数据集，表示包含行和列的数据表。

Spark结构化流

结构化流是一种基于Spark SQL引擎的可扩展、可容错的流处理引擎。通过Structured Streaming，你可以将发布到Kafka的数据视为无界DataFrame，并使用与批处理相同的DataFrame，Dataset和SQL API处理此数据。

随着流数据的不断传播，Spark SQL引擎会逐步持续处理并更新最终结果。

事件的流处理对实时ETL、过滤、转换、创建计数器、聚合、关联值、丰富其他数据源或机器学习、持久化文件或数据库以及发布到管道的不同topic非常有用。

Spark结构化流示例代码

下面是Uber事件数据聚类分析用例的数据处理管道，用于检测位置。

使用Kafka API将行车位置数据发布到MapR-ES topic

订阅该topic的Spark Streaming应用程序：

• 输入Uber行车数据流;
• 使用已部署的机器学习模型、集群ID和位置丰富行程数据;

在MapR-DB JSON中存储转换和丰富数据。

用例数据示例

示例数据集是Uber旅行数据，传入数据是CSV格式，下面显示了一个示例，topic依次为：

日期/时间，纬度，经度，位置(base)，反向时间戳

2014-08-06T05：29：00.000-07：00,40.7276，-74.0033，B02682,9223370505593280605

我们使用集群ID和位置丰富此数据，然后将其转换为以下JSON对象：

{  
"_id":0_922337050559328,  
"dt":"2014-08-01 08:51:00",  
"lat":40.6858,  
"lon":-73.9923,  
"base":"B02682",  
"cid":0,  
"clat":40.67462874550765,  
"clon":-73.98667466026531  
} 

加载K-Means模型

Spark KMeansModel类用于加载k-means模型，该模型安装在历史uber行程数据上，然后保存到MapR-XD集群。接下来，创建集群中心ID和位置数据集，以便稍后与Uber旅行位置连接。

集群中心下方显示在Zeppelin notebook中的Google地图上：

从Kafka的topic中读取数据

为了从Kafka读取，我们必须首先指定流格式，topic和偏移选项。有关配置参数的详细信息，请参阅MapR Streams文档。

这将返回具有以下架构的DataFrame：

下一步是将二进制值列解析并转换为Uber对象的数据集。

将消息值解析为Uber对象的数据集

Scala Uber案例类定义与CSV记录对应的架构，parseUber函数将逗号分隔值字符串解析为Uber对象。

在下面的代码中，我们使用parseUber函数注册一个用户自定义函数(UDF)来反序列化消息值字符串。我们在带有df1列值的String Cast的select表达式中使用UDF，该值返回Uber对象的DataFrame。

使用集群中心ID和位置丰富的Uber对象数据集

VectorAssembler用于转换并返回一个新的DataFrame，其中包含向量列中的纬度和经度要素列。

k-means模型用于通过模型转换方法从特征中获取聚类，该方法返回具有聚类ID(标记为预测)的DataFrame。生成的数据集与先前创建的集群中心数据集(ccdf)连接，以创建UberC对象的数据集，其中包含与集群中心ID和位置相结合的行程信息。

最后的数据集转换是将唯一ID添加到对象以存储在MapR-DB JSON中。createUberwId函数创建一个唯一的ID，包含集群ID和反向时间戳。由于MapR-DB按id对行进行分区和排序，因此行将按簇的ID新旧时间进行排序。 此函数与map一起使用以创建UberwId对象的数据集。

写入内存接收器

接下来，为了进行调试，我们可以开始接收数据并将数据作为内存表存储在内存中，然后进行查询。

以下是来自 %sqlselect * from uber limit 10 的示例输出：

现在我们可以查询流数据，询问哪段时间和集群内的搭乘次数最多?(输出显示在Zeppelin notebook中)

1. %sql 

SELECT hour(uber.dt) as hr,cid, count(cid) as ct FROM uber group By hour(uber.dt), cid

Spark Streaming写入MapR-DB

用于Apache Spark的MapR-DB连接器使用户可以将MapR-DB用作Spark结构化流或Spark Streaming的接收器。

当你处理大量流数据时，其中一个挑战是存储位置。对于此应用程序，可以选择MapR-DB JSON(一种高性能NoSQL数据库)，因为它具有JSON的可伸缩性和灵活易用性。

JSON模式的灵活性

MapR-DB支持JSON文档作为本机数据存储。MapR-DB使用JSON文档轻松存储，查询和构建应用程序。Spark连接器可以轻松地在JSON数据和MapR-DB之间构建实时或批处理管道，并在管道中利用Spark。

使用MapR-DB，表按集群的键范围自动分区，提供可扩展行和快速读写能力。在此用例中，行键_id由集群ID和反向时间戳组成，因此表将自动分区，并按最新的集群ID进行排序。

Spark MapR-DB Connector利用Spark DataSource API。连接器体系结构在每个Spark Executor中都有一个连接对象，允许使用MapR-DB(分区)进行分布式并行写入，读取或扫描。

写入MapR-DB接收器

要将Spark Stream写入MapR-DB，请使用tablePath，idFieldPath，createTable，bulkMode和sampleSize参数指定格式。以下示例将cdf DataFrame写到MapR-DB并启动流。

使用Spark SQL查询MapR-DB JSON

Spark MapR-DB Connector允许用户使用Spark数据集在MapR-DB之上执行复杂的 SQL 查询和更新，同时应用投影和过滤器下推，自定义分区和数据位置等关键技术。

将数据从MapR-DB加载到Spark数据集中

要将MapR-DB JSON表中的数据加载到Apache Spark数据集，我们可在SparkSession对象上调用loadFromMapRDB方法，提供tableName，schema和case类。这将返回UberwId对象的数据集：

使用Spark SQL探索和查询Uber数据

现在，我们可以查询连续流入MapR-DB的数据，使用Spark DataFrames特定于域的语言或使用Spark SQL来询问。

显示第一行(注意行如何按_id分区和排序，_id由集群ID和反向时间戳组成，反向时间戳首先 排序 最近的行)。

df.show 

每个集群发生多少次搭乘?

df.groupBy("cid").count().orderBy(desc( "count")).show 

或者使用Spark SQL：

%sql SELECT COUNT(cid), cid FROM uber GROUP BY cid ORDER BY COUNT(cid) DESC 

使用Zeppelin notebook中的Angular和Google Maps脚本，我们可以在地图上显示集群中心标记和最新的5000个旅行的位置，如下可看出最受欢迎的位置，比如位于曼哈顿的0、3、9。

集群0最高搭乘次数出现在哪个小时?

df.filter($"\_id" <= "1")  
.select(hour($"dt").alias("hour"), $"cid")  
.groupBy("hour","cid").agg(count("cid")  
.alias("count"))show 

一天中的哪个小时和哪个集群的搭乘次数最多?

%sql SELECT hour(uber.dt), cid, count(cid) FROM uber GROUP BY hour(uber.dt), cid 

按日期时间显示uber行程的集群计数

%sql select cid, dt, count(cid) as count from uber group by dt, cid order by dt, cid limit 100 

总结

本文涉及的知识点有Spark结构化流应用程序中的Spark Machine Learning模型、Spark结构化流与MapR-ES使用Kafka API摄取消息、SparkStructured Streaming持久化保存到MapR-DB，以持续快速地进行SQL分析等。此外，上述讨论过的用例体系结构所有组件都可与MapR数据平台在同一集群上运行。

代码：

你可以从此处下载代码和数据以运行这些示例：https：//github.com/caroljmcdonald/mapr-spark-structuredstreaming-uber

机器学习notebook的Zeppelin查看器：https：//www.zepl.com/viewer/github/caroljmcdonald/mapr-spark-structuredstreaming-uber/blob/master/notebooks/SparkUberML.json

Spark结构化流notebook的Zeppelin查看器：https：//www.zepl.com/viewer/github/caroljmcdonald/mapr-spark-structuredstreaming-uber/blob/master/notebooks/SparkUberStructuredStreaming.json

SparkSQL notebook的Zenpelin查看器：https：//www.zepl.com/viewer/github/caroljmcdonald/mapr-spark-structuredstreaming-uber/blob/master/notebooks/SparkUberSQLMapR-DB.json

此代码包含在MapR 6.0.1沙箱上运行的说明，这是一个独立的VM以及教程和演示应用程序，可让用户快速使用MapR和Spark。