流式计算在交通管理中应用范文

【摘要】

近年来，为强化路面动态监控，各级公安交通管理部门开始大规模建设车辆智能监测记录系统（以下简称卡口系统），随着卡口系统接入的卡口数量日益增多，当卡口的过车数据量大到一定时候，基于传统关系型数据库的比对预警方式，会出现预警时间延迟，无法满足实战的实时性需求。本文针对卡口过车的实时预警要求，对大数据流式计算组件进行了分析和研究，提出了过车信息实时接入、实时比对预警的大数据流式计算技术解决方案，为基层路面民警的及时拦截查处争取了时间。

【关键词】

大数据；流式计算；SparkStreaming；智能交通；卡口

引言

近年来，为强化路面动态监控，各级公安交通管理部门开始大规模建设车辆智能监测记录系统(以下简称卡口系统)，根据公安部“金盾工程”总体建设以及公安机关图像信息联网应用要求，以公安交通管理综合应用平台为依托，整合共享各地车辆智能监测记录系统信息资源，建立横向联网、纵向贯通的交通安全主动防控云平台，满足各级公安交通管理部门车辆缉查布控和预警拦截、车辆轨迹和交通流量分析研判、交通违法行为甄别查处等业务应用。随着各地卡口系统接入的卡口数量日益增多，基于传统关系型数据库的实时比对预警，无法满足实战的实时性的需求。自从Google了基于云计算的分布式大数据处理编程模型，大数据技术得到了广泛的应用，开源的Hadoop分布式计算软件框架更是将大数据应用推向了极限，网页搜索、精准营销等典型应用的成功使Hadoop、MapReduce成为大数据的象征。MapReduce是一种离线的批处理方式，可以成功处理TB、PB级海量数据，但无法应对实时数据分析需求和对消息事件的实时响应，大数据处理需要支持实时处理和迭代计算技术作为补充，因此流式计算成为大数据技术研究的新热点。流式计算来自于一个信念：数据的价值随着时间的流逝而降低，所以事件出现后必须尽快对它们进行处理，而不是缓存起来成批处理。基于卡口海量的实时过车信息，如何与黑名单信息快速的比对预警，成为当前主动防控云平台应用的关键技术。本文提出了基于大数据流式计算的快速比对的解决方案，实现嫌疑车辆快速比对预警，为基层路面民警的及时拦截查处争取了时间。

1主动防控平台概况

按照公安部公路交通安全防控体系建设要求，基于全国机动车缉查布控系统[1]，应用大数据、云计算技术，实现了卡口机动车过车信息汇聚，实现海量过车信息查询、轨迹分析、套牌分析[2]、伴随分析、碰撞分析、区间测速等功能，实现跨区域、跨警钟的信息共享、深度挖掘，为监测公路运行情况、快速查缉违法行为、打击涉车犯罪、提升公路安全管控水平和社会安全服务水平。

1.1软件架构

（1）如图1所示，在分布式消息总线集群服务器上构建基于Kafka的分布式消息总线，前端卡口将过车信息临时存储在Kafka消息队列中；（2）在流计算集群服务器上构建基于SparkStreaming的实时流式计算，实现过车信息与机动车登记信息、黑名单信息等实时关联分析；（3）在分布式存储集群服务器上构建基于HBase[3]的分布式数据库，实现过过车信息、流量统计信息、嫌疑车辆信息，黑名单信息等存储；（4）构建基于Hadoop大数据引擎[4]，实现关系型数据库和分布式数据库的数据关联应用和可视化展示。1.2平台功能按照面向基层，贴近实现、重在应用的工作思路，基于卡口过车信息，研发了车辆监控、缉查布控、执勤执法、分析研判等四大功能模块，具体功能如下。1.2.1车辆监控功能实现基于GIS卡口、视频实时监控、目标车辆实时追踪、机动车轨迹查询等功能。

1.2.2缉查布控功能

采用大数据流式计算技术，实现过车的实时比对预警、重点人员车辆的实时比对预警、假套牌车辆的实时比对、区间测速、流量统计的实时运算。

1.2.3执勤执法功能

实现执法服务站管理、重点车辆检查登记、现场违法非现场拦截查处等功能。1.2.4分析研判功能实现对公路客运、旅游客运、危险品运输车辆、逾期未检验、逾期未报废、凌晨2时至5时客运车辆违规上路行驶、重要路段区间测速、道路交通流量等分析功能，实现了对嫌疑假牌、套牌、伴随车辆、碰撞车辆等分析研判功能。

2数据处理架构

数据处理架构由数据采集、数据接入、流式计算、数据输出等四部分构成.数据采集：车辆智能监测记录系统的前端卡口负责过车信息采集，包括文本和图片信息。数据接入：车辆智能监测记录系统调用全国机动车缉查布控系统提供Webservice接入服务，将过车信息写入，过车信息使用Kafka分布式消息队列作为缓冲，接入服务不再负责比对。流式计算：使用Storm或SparkStreaming等流式计算技术，从Kafka分布式消息队列中取数据进行实时比对处理。数据输出：比对结果通过JDBC方式输出至关系型数据库ORACLE。

2.1流式计算技术

流式计算技术和批量处理技术有着本质的差别，流式计算技术需要维护消息队列并进行实时消息的及时处理。分布式流式计算技术虽然处于起步发展阶段，但由于市场广泛需求的驱动，成为关注和研究热点。当前具有代表性的流式计算技术有Storm、SparkStreaming[2]。Storm是Twitter支持开发的一款分布式、开源的、实时的、高容错的大数据流式计算系统。Storm集群主要由一个主节点和一群工作节点构成，通过Zookeeper进行协调。如图3所示，在Storm中，先要设计一个用于实时计算的图状结构，我们称之为拓扑（topology）。这个拓扑将会被提交给集群，由集群中的主控节点（masternode）分发代码，将任务分配给工作节点（workernode）执行。一个拓扑中包括spout和bolt两种角色，其中spout发送消息，负责将数据流以tuple元组的形式发送出去；而bolt则负责转换这些数据流，在bolt中可以完成计算、过滤等操作，bolt自身也可以随机将数据发送给其他bolt。由spout发射出的tuple是不可变数组，对应着固定的键值对[5,6]。如图4所示，SparkStreaming是核心SparkAPI的一个扩展，它并不会像Storm那样一次一个地处理数据流，而是在处理前按时间间隔预先将其切分为一段一段的批处理作业。Spark针对持续性数据流的抽象称为DStream（DiscretizedStream），一个DStream是一个微批处理（micro-batching）的RDD（弹性分布式数据集）；而RDD则是一种分布式数据集，能够以两种方式并行运作，分别是任意函数和滑动窗口数据的转换。表1给出了Storm、SparkStreaming的功能，性能等对比，基于下述对比，选择了SparkStreaming流式计算技术。

2.2分布式消息队列

Kafka也是Apache[7]下的开源消息系统项目，是一种高吞吐量的分布式消息订阅系统，在普通的服务器上每秒也能处理几十万条消息，可用于低时延的收集和发送大量的事件和日志数据。Kafka也是Apache下的开源消息系统项目，是一种分布式的，基于/订阅的消息系统。它以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能。具有高吞吐量，即使在非常普通的硬件机器上也能做到单机支持每秒十万条以上消息的传输。支持KafkaServer间的消息分区及分布式消费，同时能保证每个Partition内的消息顺序传输。同时支持离线数据处理和实时数据处理，并且支持在线水平扩展。Kafka包括以下四个组件：一是话题（Topic），它是特点类型的消息流，消息是字节的有效负载，话题是消息的分类名；二是生产者（Producer），它是能够消息到话题的任何对象；三是（Broker）或Kafka集群，已的消息保存在其中；四是消费者（Consumer），它可以订阅一个或多个话题，并从拉取数据，从而消费这些已的消息。Kafka的整体架构如图5所示。因为Kafka内在就是分布式的，一个Kafka集群通常包括多个。为了均衡负载，将话题分成多个分区，每个存储一或多个分区。多个生产者和消费者能够同时生产和获取消息。

2.3SparkStreaming与Kafka集成

Kafka[6]是一个分布式的消息-订阅系统，下面介绍如何使用SparkStreaming从Kafka中接收数据，具体包括两种方法：一是使用Receivers和Kafka高层次的API；二是使用DirectAPI，这是使用低层次的KafkaAPI，并没有使用到Receivers，是Spark1.3.0中开始引入的。

2.3.1基于Receivers的方法

这个方法使用了Receivers来接收数据。如图6，Receivers的实现使用到Kafka高层次的消费者API。对于所有的Receivers，接收到的数据将会保存在Sparkexecutors中，然后由SparkStreaming启动的Job来处理这些数据。然而，在默认的配置下，这种方法在失败的情况下会丢失数据，为了保证零数据丢失，你可以在SparkStreaming中使用WAL日志，这是在Spark1.2.0才引入的功能，这使得我们可以将接收到的数据保存到WAL中（WAL日志可以存储在HDFS上），所以在失败的时候，我们可以从WAL中恢复，而不至于丢失数据。

2.3.2基于DirectAPI的方法和基于Receiver

接收数据不一样，这种方式定期地从Kafka的topic+partition中查询最新的偏移量，再根据定义的偏移量范围在每个batch里面处理数据。当作业需要处理的数据来临时，spark通过调用Kafka的简单消费者API读取一定范围的数据。如图7和基于Receiver方式相比，这种方式主要有几个优点：（1）简化并行。我们不需要创建多个Kafka输入流，然后union他们。而使用directStream，SparkStreaming将会创建和Kafka分区一样的RDD分区个数，而且会从Kafka并行地读取数据，也就是说Spark分区将会和Kafka分区有一一对应的关系，这对我们来说很容易理解和使用。（2）高效。第一种实现零数据丢失是通过将数据预先保存在WAL中，这将会复制一遍数据，这种方式实际上很不高效，因为这导致了数据被拷贝两次：一次是被Kafka复制；另一次是写到WAL中。（3）恰好一次语义（Exactly-oncesemantics）。通过Kafka低层次的API，并没有使用到Zookeeper，偏移量仅仅被SparkStreaming保存在Checkpoint中。这就消除了SparkStreaming和Zookeeper中偏移量的不一致，而且可以保证每个记录仅仅被SparkStreaming读取一次，即使是出现故障。

3流式计算解决方案

随着前端卡口接入数量的不断增加，过车数据规模的不断扩大，使用传统的逻辑架构会造成以下两个问题：一是过车数据上传积压问题，传统的传输机制已不能满足大数据量的过车信息上传；二是实时比对效率降低问题，通过接入服务程序提供的Webservice或Servlet接口，实现过车信息接入，接入时进行比对预警，当数据量大的时候，无法及时预警。基于以上问题我们采用大数据库流式计算技术，使用Kafka分布式消息总线作为缓冲，接入服务不再负责比对，只负责提供接口写入数据至Kafka，然后由SparkStreaming从Kafka中取数据进行实时比对预警，并将结果输出到交通安全主动防控平台中。

3.1比对预警示意图

3.2过车等9种信息接入

通过接入服务器过实现过车信息、流量检测信息、气象检测信息、交通事件信息、交通诱导信息、停车场车辆停车信息、警车定位信息、警员定位信息、非现场违法信息9种数据接入，Kafka以Topic来进行消息管理，在系统中按每一数据类型设定相应的Topic，然后由相应的Consumer去负责消费需要的Topic数据。

3.3基础信息内存加载

为了更快的信息加载速度，系统先定期将机动车登记信息、黑名单信息装载至HBase分布式数据库，然后SparkStreaming再从分布式数据库加载机动车登记信息、黑名单信息。SparkStreaming信息加载时分为全项信息加载和根据hash算法部分信息加载。其中全项信息的加载由后台任务定时加载；根据hash算法的加载由Consumer任务在拉取partitions数据时触发加载根据partitions的hash算法决定要加载那部分车辆的基础信息和布控车辆信息（目前沿用山东项目的算法，根据号牌号码信息和partitions的个数）。

3.4比对预警信息生成

在系统中，对于从Kafka中实时获取到的Topic数据，SparkStreaming作为Consumer负责动静态信息的实时碰撞、分析和预警，区间测速、旅行时间计算等。系统已实现人工布控黑名单信息、机动车登记信息、驾驶人管理信息、全国客运车辆和危险品运输车、逾期未年检、逾期未报废等重点车辆信息数据和卡口过车信息实时碰撞分析，对嫌疑车辆在秒级发出实时预警信息，指挥中心民警在接收到预警信息后可及时指挥路面民警对嫌疑车辆进行拦截查处。流式计算的最终结果，对于海量的布控黑名单轨迹信息、重点车辆轨迹信息等根据业务类型存放到HBase中相应的业务表中，对于预警信息，存放到Oracle关系型数据库，便于后续业务处理。

4结论

大数据流式计算和批量计算适用于不同的应用场景。批处理汇聚海量数据分析出的结果可能更精确，但对数据时效性要求严格而对历史数据积累并不非常关注的场景，流式计算具有明显的优势。批量计算和流式计算是有优劣互补的，因此在多种应用场合下可以将两者结合起来使用。目前，山东全省已实现了1023套卡口上传过车信息100毫秒内接收并预警，日接入过车信息超过1550万。实现了各类动、静态信息的实时比对，嫌疑车辆300毫秒内发出预警。与传统采用关系型数据库相比，采用SparkStreaming流式计算技术的比对预警更快、监测的种类更多。山东已实现逾期未检验、逾期未报废、强制注销、车主驾照满分、暂扣等实时比对预警，为基层民警的应用提供了数据支撑。通过采用SparkStreaming流式计算技术，解决了数据的积压问题、保证了业务数据的有效性，解决前端卡口接入性能、保证了比对预警时效性、从而大大提高了交管部门的管控能力。基于Hadoop的大数据云计算平台扩展性强，存储和计算能力可以不断提升，充分运用大数据云计算技术，让交通管理变得更加“智慧”。

作者：周建宁 徐晓东 蔡岗 单位：公安部交通管理科学研究所