[参赛项目] 基于Flink流处理框架的交通灯控制器

项目简述

本项目旨在基于分布式系统架构下处理复杂交通灯系统。由多组件组成，这些组件进行通信以确保交通灯按正确顺序打开，从而最大限度地减少车辆驾驶时的交通拥塞时间浪费，同时系统必须反馈有关交通及交通路口的统计数据。本项目的整个数据流是在交通灯附近行驶用户利用Kafka主题发送数据，信号量本身计算有关车辆的记录统计信息并向它们发送另一个主题。本项目的数据由Flink Dispatcher 和 semaphoreController两者读取。第一个组件使用消息根据通过交通路口的车辆速度的平均值和中值计算排名。semaphoreController 使用 Flink 读取的数据作为 Q-Learning 算法的输入。变绿的信号量是基于该算法决定的。由于这个阶段，用户浪费的时间被最小化。整个系统状态保存在 MongoDB 上。用户可以使用 Web 应用程序控制查询结果和系统活动节点。

背景

目前城市交通拥堵问题日益严峻，产生了很大的社会成本，部分大城市的平均通勤时间超过 1.5 小时，如何帮助人们解决通勤的体验已经越来越受到关注。由于道路交通状况瞬息万变，因此实时计算在这方面有天然的用武之地。同时越来越多的设备通过互联网相互连接，允许它们之间进行强大的控制，并且可以收集大量信息，有助于提供更有效的服务。

目标

本项目中我们将利用这项创新改进实时交通灯提供的服务，使其能够提供实时交通状况流量更改交通灯时间，同时建立基于分布式机制来控制交通灯时间，同时收集危险交通灯路口的有用信息。

实施方案

系统结构

假设目前系统的每个信号量都在正常执行任务，同时传感器在不断注册信息上传至云端同时控制器接受使用过程收到的记录，允许根据实时方向的流密集度，自适应调整允许对应的交通灯保持红-绿灯间隔状态时间。

png1

SemaphoreController 模块管理系统中与信号量的收集相关功能,由于Semaphore不是唯一能够发送信息的载体；在智能手机等智能设备已成为普遍的设备，利用它们的传播和网络链接来获取越来越重要的数据是友好的方式所以SemaphoreController模块的Semaphore具有扩展性同时SemaphoreController模块在信号量处理并将通知用户，同时利用机器学习算法控制交通灯路口对信号量进行计时。设备间通信出现问题，SemaphoreController模块会建立紧急模式。
如果不是所有载体都收到一条消息就会出现不愉快的情况，基于此使用2PC协议来交换消息，将事务提交过程分成两个阶段来进行处理，使得每个载体必须知道控制器做出的决定。
传感器发送的数据被一个组件接收，该组件在执行流处理、聚合并输出有效的统计数据以检索。这样的所有流程都由发布/订阅消息框架捆绑在一起，允许组件之间的简单有效消息交换。

项目结构

主要流程周期：
1. 数据由传感器生成，这些传感器生成包含不同信息的元组流。这些元组被生成到kafka主题中，从Flink和Java实例中读取。Flink使用这些数据来计算聚合并发送到Kafka主题的结果。这些结果由名为Monitor的Java实例读取，以便计算查询，计算结果被写入基于分布式文件存储的数据库MongoDB。
2. 传感器元组与发送数据的信号量控制器的读取流，由于信息号量控制器可以决定那个信号量元组可以设置为绿色，这决定是在不同Kafka主题上发送的，信号量可以在这些主题上收听并被告知那些状态必须改变。
主要模块功能如下： Fornt end：由一个网页与监视器实例组成。检索有关信息的唯一方法是从数据库中读取。 IoTSimulator：使用Kafka代理模拟IOT数据交换。 Semaphore：使用Kafka读取交通灯路口信息并发送传感器数据并提供REST接口在实例化期间由用户进行配置。 Semaphore Controller：涉及2PC协议在信号量中交换消息，检索由机器学习算法做出决策或包括管理交通灯在内的一切。 Crossroad controller：此实例可以访问存储系统并存储它负责的信号量状态以及自身状态，允许添加新的信号量或删除它们。 Monitor：它从 Kafka 读取 Flink 结果，计算查询并将输出存储在 MongoDB 上。 Monitor front end：从 MongoDB 读取数据并提供用户界面来检查十字路口的信号量。 Monitor Back end：在REST接口上接收用户请求。 Flink Dispatcher：从 Kafka 主题读取传感器数据并使用 Apache Flink 进行流处理，发送关于另一个 Kafka 主题的结果。

数据产生和消息逻辑

数据生成:数据由传感器生成，这些传感器生成包含不同信息的元组流。这些元组被生成到kafka主题中，从Flink和Java实例中读取。Flink使用这些数据来计算聚合并发送到Kafka主题的结果。
数据内容：信号量的真实数据集（foreach的纬度、经度）传感器（信号量和移动设备）产生的数据是虚构的，使用Java内置的伪随机数生成器库生成连续随机的数据。信号量传感器元组包含与信号量本身和观察到的汽车流量（单位时间内的汽车平均速度）相关数据。移动设备记录关于接近信号量的汽车当前信息，这意味着它的平均速度和坐标等。
数据注入：将这些记录数据发送到相应的Kafka主题中进行进一步分析。主题中的记录由控制器使用以点亮信息量或由Flink使用以计算查询。每个查询引用不同的主题。Flink制作的元组或任何必须共享的信息，都封装在一个Message类中主要目的是将代码关联到记录中以供过滤使用。

数据处理和查询计算

发送到Flink实例的输入记录采用JSON格式，因为JSON格式是最广泛、最常用、最容易理解和最容易生成的格式。每当读取JSON字符串时，自定义反序列器floatmap将其转换为Tuple x，将成为分析流的一部分。
利用Flink的aggregate（）方法，使用聚合器元组执行迭代计算。尽管迭代计算平均值很容易但对于中位数等仍然需要获取一个近似值，所以使用拟合优度来观测值的拟合程度，使用PSquared 算法达到这样的目的。
Flink在TimeWindow建立时间序列长度计算，然后输出结果，同时保持聚合新数据。输出每15分钟、1小时、24小时生成一次，并生成序列化JSON格式并放置在Kafka Topic中。

解决查询的机器学习算法

在开始本项目之前，检索智能交通灯在城市中实施的科学论文中均都使用类似相同的强化学习方法：Q-Learning，Q-Learning是强化学习算法中value-based的算法，Q即为（s,a）就是在某一时刻的s状态下(s∈S)，采取采取动作a (a∈A)动作能够获得收益的期望，环境会根据agent的动作反馈相应的回报reward r，所以算法的主要思想就是将State与Action构建成一张Q-table来存储Q值，然后根据Q值来选取能够获得最大的收益的动作。

Q-Learning的目标是学习一个策略，抽象为智能体（Agent）、环境状态（environment）、奖励（reward）、动作（action）下告诉代理什么情况下采取什么行动。由于Q-Learning需要一个奖励函数，简单起见我们对每个信号量的队列进行了估计。Q-Learning必须回答： 1、为什么要打开红绿灯的必要性？ 2、一直关闭红绿灯的危害性？每个信号量实体包含两个值，每个值都是关于之前提出问题的指标。通过比较每个信号量的值选择最适合开启的且学习公式如下：

Q(k+1)(state(k+1),learning_rate) =(1 − α)Qk(state, learning_rate)+α(queue_size + γ + mawQ)

queue_size参数是通过放置在信号量中的传感器注册的信息获得的。当前状态每个用户定义的时间段更新，随着时间的推移允许更精准的决策。

紧急模式机制

1、当交通灯中断或者节点链接丢失会采取的什么方案？紧急模式是种协议，可以确保只有一个信号量被设置为绿色，并且可以在缺乏链接的情况下避免交通事故的发生。因为即使出现问题或不再收集数据，也可以高效的管理流量。在发生传感器故障会让Flink发送消息困难，紧急模式允许稳定的信号量工作。紧急模式是通过定义一个循环时间来实现的，该时间给出了关于绿色信号量在红色信号量占比多少度量，因为强调了交通灯坏掉后的事件的严重性，因此考虑实施实时 2、通知系统的机制，可以直观且不同于最常用的解决方案。构建交通灯管理系统中创建仅管理员可以读写的组，这样可以避免垃圾信息并且只能发送紧急消息，附加损坏的信号量ID，地理坐标并最终损坏的灯泡类别。这种解决方案提高了与管理人员的可视化交互能力，无需担心创建完整系统的紧急状况。

数据存储格式:

在架构结构所解释的一样，用户必须有权访问系统的状态，而无需与服务端进行通信，所以就需要存储整体系统结构。
三级结构中每级项目都有一个唯一的ID和一个类型属性。类型属性用于更简单的方式查询数据库。
每个交通路口都有一个用于MongoDB中进行查询的ID和另一个以交通灯路口ID命名的用于后端和前端的ID。
信号量有一个ID，该ID被分配计算其纬度和经度的哈希函数SHA256以及一个整数故障用来表示信号量识别损坏灯泡的次数。
每个信号控制器都以唯一ID为特征，并由类型和包含受其控制的一组交通灯路口的数组定义。数组中包含的每个交通灯路口都由其他属性定义，如ID、street、类型字段、信号量列表等。
MongoDB也用于保存查询结果，schema表示结构。系统的每个查询都使用相同的模式，并且查询具有唯一标识符和称为排名的列表属性，该属性按顺序包含所有对象且每个对象都由唯一标识符标识。
Flink计算的结果是通过比较值数据计算出的值，其余属性是向前端提供更快的信息。

成员介绍

阿里云天池昵称：6utdmwrvoyumo

flink-china / flink-forward-asia-hackathon-2021