基于SparkStreaming的车辆轨迹实时处理实现研究

马晟 刘雅伦 陈晓男 沈漪

摘  要：城市的发展使得运营车辆日益增长，车辆调度愈发困难，传统系统无法满足现有众多车辆的监控调度与运营。该程序基于大数据流处理系统，实现了大批量的车辆信息监测和实时处理以及车辆的精细监控与轨迹回放。可用于网约车、公交车以及货运集团的调度中心进行实时监控和订单把控，以提高车辆调度的灵活性，达到最优调度、减少成本的效果。

关键词：SparkStreaming  大数据  轨迹回放  交通

中图分类号：TP31       文献标识码：A

Abstract： With the development of the city， the number of operating vehicles is increasing， and the vehicle scheduling is becoming more and more difficult. The traditional system can not meet the monitoring， scheduling and operation of many existing vehicles. Based on the large data stream processing system， the program realizes a large number of vehicle information monitoring and real-time processing， as well as vehicle fine monitoring and track playback. It can be used for real-time monitoring and order control in the dispatching center of online car hailing， buses and freight groups， so as to improve the flexibility of vehicle scheduling， achieve optimal scheduling and reduce costs.

Key Words：SparkStreaming;Bigdata;Track playback;Traffic

隨着城市的发展，运营车辆日益增长，车辆调度愈发困难，传统系统无法满足现有众多车辆的监控调度与运营。基于大数据系统的车辆实时监控与调度需求随着大数据技术的日趋发展有了实现的可能。

1  数据处理系统的设计

该系统实现对海量车辆轨迹数据的采集、存储、实时处理、轨迹回放功能。轨迹数据在盖亚数据平台申请达到，编程模拟产生实时数据流，经大数据平台采集处理存入数据库，然后在前台显示实时的车辆轨迹[1]。

1.1系统整体架构

基于系功能需求，该系统的总体设计为：先由车辆端上传坐标数据（编程模拟产生），flume多源采集，然后写入kafka的topic，接着通过SparkStreming实时消费kafka，再根据订单存入redis，最后实现订单数据列表生成以及订单车辆轨迹回放[2]。它异于传统数据系统的地方是：采用大数据流处理框架，具有高吞吐率、高负载、高可用性、实时性高的优点[3]。整个系统的逻辑实现如图1所示。

1.2数据回放模块设计

为了模拟真实业务场景，该程序基于盖亚平台坐标数据通过数据回放模块模拟数据流产生[4]。使用python读出坐标数据，用多线程并行输出，从而模拟实际场景中车辆移动汇报的坐标打点数据，达到采集流数据的需求[5]。

核心代码逻辑如下所示。

#坐标数据文件写入

def consumer（queue， writer， csv_file）：

while True：

line = queue.get（）

deal_line（line， writer， csv_file）

queue.task_done（）

#流数据文件生成

def producer（queue）：

with open（‘test.txt’， ‘r’） as f：

for line in f：

queue.put（line）

queue = JoinableQueue（8）

pc = Process（target=producer， args=（queue，））

for _ in range（cpu_count（））：

c1 = Process（target=consumer， args=（queue， writer， csv_file））

#等待生产者进程全部生成完毕

pc.join（）

#等待所有数据全部处理完毕

queue.join（）

1.3数据采集消费模块设计

该模块实现了通过flume采集车辆轨迹流数据，进而推送到消息队列kafka中。

首先进行flume数据采集，在采集过程中通过集群形式达到大数据量及多源数据采集情况下的负载均衡及并行采集。设置flume静态拦截器实现在采集到的数据的头数据中插入自定义的key-value键值对以区分不同数据源，主要配置如下：

a1.sources.r1.interceptors.i1.type = static          #设置静态拦截器

a1.sources.r1.interceptors.i1.key = type

a1.sources.r1.interceptors.i1.value = test_gps_topic#不同的数据源取不同的名称

接着通过kafka集群接收flume采集的大量数据，以实现数据高吞吐率、高可用数据传递以及数据的实时处理，同时通过不同的topic保证不同数据流的分区。

flume监听的文件数据发送到此kafka的主题当中，主要配置如下：

a1.sinks.k1.topic = test_gps_topic           #与前面的静态拦截器value值配置相一致

1.4数据实时处理模块设计

该模块通过sparkStreaming程序实现消费kafka中的数据存到HBase中，其中的GPS位置经纬度信息保存到redis中，存为后续实时监控以及轨迹回放的数据源[6]。核心逻辑的Scala代码如下：

//从kafka里消费数据，把经纬度信息存到redis

val result： InputDStream[ConsumerRecord[String， String]] = Tools.getStreamingContextFromHBase

（streamingContext，kafkaParams，topics，group，"（.*）gps_topic"）

result.foreachRDD（eachRdd =>{

eachRdd.foreachPartition（eachPartition =>{

val connection： Connection = HBaseUtil.getConnection

val jedis： Jedis = JedisUtil.getJedis

eachPartition.foreach（record =>{

Tools.saveToRedis（connection，jedis，record）

}）

1.5轨迹回放模块设计

得益于redis内存数据库高性能以及可持久化的稳定性，该模块实现回放每个订单车辆轨迹同时并发实时读取到前端，通过高德地图提供的地图api接口，订单号为同一个key的value坐标数据轨迹点按时间顺序呈现在地图上，从而监控每条车辆订单的车辆轨迹情况[7]。

2  实验验证

2.1 實验环境

该次实验采用了一主二从的CDH集群，机器配置如图2所示，集群角色配置如图3所示。

2.2数据集

此次实验数据集来自滴滴盖亚数据平台的开放数据，形如表1所示。

首先是数据回放模块的验证，通过多线程输出，flume采集源目录，数据如期以多订单并发每秒三条的流数据形式生成。其次是数据实时处理模块，经检查redis数据库，回放的流数据以秒级单位处理写入到数据库。最后是数据回放模块的验证，经前端程序的读取，车辆轨迹坐标成功呈现在了高德地图上。

3 结语

该系统实现了大规模轨迹数据的处理，数据的吞吐量、延迟性、精准度已达到预期。程序通过Python模拟车辆轨迹数据流的产生，然后通过flume和kafka采集消费数据，sparkStreaming处理数据流，完成了模拟现实生活多车辆多数据流场景的数据产生、处理与轨迹回放。目前程序还停留在雏形阶段，未来将在耦合度、灵活度上做出提高。

参考文献

[1]  杨小润.基于深度学习的车辆轨迹特征识别与分析[D].南京：南京邮电大学，2020.

[2]  陆键，王可，蒋愚明.基于车辆行驶轨迹的道路不良驾驶行为实时辨识方法[J].交通运输工程学报，2020，20（6）：227-235.

[3] 潘伟博，汪海涛，姜瑛，等.Hadoop集群异常节点实时检测与诊断算法[J].陕西理工大学学报：自然科学版，2021，37（4）：24-31.

[4]  鲍裕麟.深度学习应用场景下的HDFS性能优化[D].合肥：中国科学技术大学，2021.

[5] 谢枫，娄静涛，赵凯，等.基于行为识别和曲率约束的车辆轨迹预测方法研究[J].汽车工程，2019，41（9）：1036-1042.

[6] 柯杰.基于SparkStreaming日志实时监测系统的设计与实现[D].南京：东南大学，2017.

[7] 苗莉.大数据云计算环境下的数据安全[J].科技资讯，2021，19（2）：31-33.