使用Spark机器学习，流技术和Kafka API监视实时Uber数据（第2部分）

这篇文章是该系列文章的第二部分，我们将建立一个实时示例，用于分析和监视Uber车载GPS行程数据。如果您尚未阅读本系列的第一部分，则应先阅读。

第一篇文章讨论了使用Apache Spark的K-means算法创建机器学习模型，以基于位置对Uber数据进行聚类。第二篇文章将讨论将保存的K-means模型与流数据一起使用，以对Uber汽车在何时何地进行集群进行实时分析。

用例示例：地理上集群的车辆/物品的实时分析

使用Kafka API将 Uber行程数据发布到MapR Streams主题。订阅该主题的Spark流媒体应用程序使用k-means模型使用与位置对应的集群ID丰富数据，并将结果以JSON格式发布到另一个主题。订阅第二个主题的Spark流媒体应用程序实时分析JSON消息。

用例数据示例

示例数据集是Uber行程数据，您可以在本系列的第1部分中了解更多信息。传入的数据记录为CSV格式。

示例行如下所示：

丰富的数据记录为JSON格式。示例行如下所示：

Spark Kafka消费者生产者代码

解析数据集记录

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

加载K均值模型

Spark KMeansModel类用于加载拟合在历史Uber行程数据上的已保存K均值模型。

模型clusterCenters的输出：

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

Spark流代码

这些是Spark Streaming Consumer Producer代码的基本步骤：

配置Kafka Consumer Producer属性。
初始化Spark StreamingContext对象。使用此上下文，创建一个从主题读取消息的DStream。
应用转换（创建新的DStreams）。
将消息从转换的DStream写入主题。
开始接收数据并进行处理。等待处理停止。

我们将通过示例应用程序代码完成所有这些步骤。

配置Kafka Consumer Producer属性

第一步是设置KafkaConsumer和KafkaProducer配置属性，稍后将使用它们来创建DStream，以接收/发送主题消息。您需要设置以下参数：

键和值反序列化器：用于反序列化消息。
自动偏移量重置：从最早或最新的消息开始读取。
引导服务器：由于代理地址实际上未由MapR流使用，因此可以将其设置为虚拟host：port。

有关配置参数的更多信息，请参见MapR Streams文档。

初始化Spark StreamingContext对象。

如下所示，ConsumerStrategies.Subscribe用于设置主题和Kafka配置参数。我们将KafkaUtils createDirectStream方法与StreamingContext，消费者和位置策略一起使用，以根据MapR Streams主题创建输入流。这将创建一个DStream来表示传入的数据流，其中每个消息都是一个键值对。我们使用DStream映射转换来创建带有消息值的DStream。