G7在实时计算的探索与实践

做者: 张皓

G7业务快览

G7主要经过在货车上的传感器感知车辆的轨迹、油耗、点熄火、载重、温度等数据，将车辆、司机、车队、货主链接到一块儿，优化货物运输的时效、安全、成本等痛点问题。

整个数据是经过车载的传感器设备采集，好比公司的Smart盒子，CTBox盒子，油感设备，温度探头等，将车辆数据上报到后端平台，在后端平台计算和处理，最后展现到用户面前。

G7的业务场景是典型的IoT场景：

• 传感器数据
• 数据种类多
• 数据质量差
• 数据低延迟
• 数据量大

其中，数据质量差的缘由是整个链条会很是的长，从传感器采集的车辆的数据，经过网络运营商将数据上报到后端服务器，再通过解析，mq，过滤，调用三方接口，业务处理，入库，整个过程很是的长，形成数据在传输过程当中出现数据重复，数据缺失等。另一点，IoT场景须要数据传输的延迟很是低，好比进出区域报警，当车辆进入到某个电子围栏中的时候须要触发报警，这个时候须要快速产生报警事件，一般不能超过30s，不然时间太长车辆已经经过了某个电子围栏区域再报警就没有价值了。再一个，数据量也是很是大的，如今天天产生轨迹点20亿+，天天产生数据量100亿+，对计算性能的要求很是高。

实时计算选型

从上面的场景咱们能够感知到，在G7的IoT场景须要的是一个低延迟，处理速度快的实时计算引擎。最开始咱们的一些架构是基于Lambda架构的，好比轨迹点计算，会使用实时计算引擎计算出实时数据，这份数据延迟比较低，可是数据不是很准确，另外须要用离线批量再计算一遍，这份数据一般比较准确，能够用来修复实时数据。这样作的缺点也比较明显，一是程序须要维护两套代码：实时程序和离线程序，二是实时数据不许确，准确的数据延迟又过高。后来咱们惊喜的发现一种基于实时处理的架构体系Kappa。

Kappa的架构是强调数据的实时性，为了保证数据的实时性有些延迟太多的数据它会建议丢弃，全部的计算逻辑只有在实时计算中，整个计算只有一套逻辑，数据从MQ中获取，通过数据处理层计算和加工，最后落入到数据存储层，对外提供数据查询功能。相对Lambda架构，Kappa架构更加适合IoT领域。

针对Kappa架构，咱们对行业主流的实时流计算框架进行了对比：

分别对主流的流计算框架：Storm，Storm Trident，Spark Streaming，Google Cloud Dataflow，Flink作了对比。基于微批量的Spark Streaming和Storm Trident延迟比较高，从这点就不适合咱们的场景。Storm的延迟很低，可是数据一致性是At Least once，容错机制比较复杂，流控会比较抖动，这些方面都是不太适合。其中，Flink的一致性保证（1.4版本还支持了end-to-end一致性），延迟比较低，容错机制的开销是比较小的（基于Chandy-Lamport的分布式快照），流控是比较优雅的（算子之间的数据传输是基于分布式内存阻塞队列），应用逻辑与容错是分离的（算子处理和分布式快照checkpoint），基于以上咱们认为flink是比较适合IoT这个场景的。

G7业务应用案例 Flink目前在G7的应用场景，主要有三方面：

• 实时计算
• 实时ETL
• 统计分析

下面分别介绍下以上三个场景的使用。

实时计算

在G7的场景中，有不少业务都属于实时计算的范畴，好比进出区域事件，超速事件，怠速事件，超速事件，疲劳报警事件，危险驾驶报警，油耗计算，里程计算等。其中疲劳报警计算是最先开始尝试使用flink来落地的。

疲劳报警业务模型

这是G7针对客户推出的G7大屏，其中风险相关的部分是根据疲劳计算得出。

根据G7的大数据计算，由于疲劳驾驶形成货车事故的比重占到整个事故的20%。对疲劳驾驶进行报警和预警就显得特别重要，能够有效下降事故发生的可能性。

根据车辆行驶的里程，驾驶员行驶的里程，驾驶时长，判断是否存在疲劳驾驶。若是超过报警阀值则报警，若是在报警阀值下面在预警阀值上面则预警。报警和预警都是下发语音到货车驾驶室提醒司机。

这个业务场景中面临的最大挑战是实时性，稳定性。只有用最短的时间、最稳定的方式将告警下发到相关人员才能最大程度减小风险。

业务流程

在整个处理流程中，首先会去获取疲劳配置，根据车辆的状态信息和司机打卡信息与疲劳配置结合，判断是否出现预警和报警。计算过程当中会把疲劳驾驶开始的状态缓存起来，疲劳驾驶结束的时候获取以前的状态数据，匹配成功以后会生成一条完整的疲劳事件。中间会调用一些接口服务好比dubbo获取车辆的配置数据、状态数据，产生的疲劳报警则会调用下发语音的接口，疲劳事件结果也会存储到hbase、mysql、kafka等。

Streaming模型

最后开发成Flink的程序，从头到到尾分别由如下算子构成：消费kafka算子、类型转换算子、数据过滤算子、异步调用第三方接口算子，窗口排序算子，疲劳处理业务逻辑算子，数据入库算子组成。

这个过程，也是踩了很多坑，咱们也有一些心得体会：

1. 算子表达尽可能单一
2. 每一个算子尽可能内聚，算子间尽可能低耦合
3. 算子打散，异步+多线程的性能发挥更好
4. 单独设置每一个算子单元的并行度，性能更优
5. hash和balance根据状况选择：只有须要使用keyby和valuestate地方才使用hash从新分布数据。其余地方尽可能使用balance而且上下游并行度一致，会将task串联成一个线程，不会走网络IO性能更高
6. 使用Asynchronous I/O 调用dubbo接口，zuul，db，hbase等外部接口

实时ETL

有部分场景是数据简单采集、处理，入库，也就是实时ETL，包括从Kafka采集数据到HDFS、DB、HBase、ES、Kafka等，这部分工做能够抽象成Flink的算子表达：Source -> Transformation -> Sink。

这部分一般能够FlinkKafkaConumser、MapFunction、JDBCAppendTableSink这类代码。以下：

统计分析

有部分场景须要有一些实时的统计分析，好比统计最近一小时内全国各城市，车辆总数，司机总数，疲劳事件，进出区域事件，打卡次数，点熄火事件等。这种场景，一般可使用Flink SQl的作实时分析，sql+窗口函数（固定窗口，滑动窗口）。代码大体以下：

实时计算平台开发和现状

在业务上的成功落地，咱们也但愿能把打造一个实时计算平台，服务各条业务线，通过差很少3个月的打磨，内部代号为Glink的实时计算平台上线，大体的架构以下：

Glink主要由如下部分组成：

1. HDFS分布式文件系统。用来存储flink任务中产生的checkpoint/savepoint数据，任务报、第三方依赖包的存储和分发，任务运行中产生的临时数据等；
2. Yarn统一计算资源平台。用来提供统一的分布式计算资源平台，任务提交，任务调度，任务执行，资源隔离功能。目前全部的flink任务都是经过yarn进行统一的计算资源管理；
3. 性能监控AMP工具。使用点评开源的Cat，在此基础上作二次开发并取名“天枢系统”。能够提供程序的耗时9五、99线、平均耗时、最大耗时、java GC监控、线程监控、堆栈信息等；
4. 集群监控管理。机器资源监控使用zabbix，提供cpu、内存、磁盘io、网络io、链接数、句柄监控。集群资源监控和管理使用开源Ambari，提供自动化安装、配置、集群总体任务、内存、cpu资源、hdfs空间、yarn资源大小监控报警；
5. 任务监控报警。使用flink提供的statsD reporter将数据上传导时序数据库InfluxDB，经过扫描Infludb数据绘制出task的处理流量，经过监控流量阀值低于预期值报警；
6. 诊断调试。使用成熟的日志查询系统 es+logstash+kibana，经过采集每一个节点的日志写入到es中， 能够在kibana中查询关键信息获取日志内存，提供诊断和调优程序的线索；
7. Flink APP 程序应用层。具体开发的flink应用程序，一般解决实时etl，统计分析，业务计算的场景；
8. Glink任务管控平台。将如下的功能进行封装，提供统一的任务管理，运维管理功能。

实时计算平台展现-任务管理

实时计算平台展现-日志和性能监控

平台的部分功能介绍：

1. 任务管理功能。提供任务发布，修改，升级，中止，申请资源，资源审核，启动日志查看功能；
2. 运维管理功能。提供日志查看，程序监控，任务监控，流量监控，异常报警等功能。

以上Glink实时计算平台的功能，基本上知足用户独立完成从程序开发，发布，调优，上线，运维的工做。

Glink-Framework开发框架

除了提供相应的平台功能，还须要在flink的生态上提供比较好的封装和工具类，所以咱们提供了开发工具的脚手架：Glink-Framework框架。

Glink-Framework提供如下封装：

1. 简化pom文件，减小大量的依赖、插件配置；
2. 三方调用集成：dubbo，zuul；
3. 三方数据库集成：mysql，redis；
4. 多环境管理；
5. 依赖版本管理；
6. 代码监测工具：checkstyle，pmd，findbugs。

平台与业务方BP合做方式

另一方面，咱们认为flink是有必定的技术门槛，特别对于以前没有并发编程、集群开发经验的小伙伴，须要有一段时间的学习才能上手，针对这个痛点，咱们提出了技术BP的技术合做方式。咱们会根据业务的复杂度，平台指派一至多名技术人员参与到业务方的整个开发和运维工做中，从需求分析到上线落地全程参与，后期还会有持续的技术分享和培训帮助业务方学习开发能力。

踩坑

在整个平台化，以及业务开发的过程当中，flink也踩坑很多，比较典型的下面一些。

1. 并行度太多形成barrier对齐的花费时间更长，有个并行度28的子任务的对齐时间超过50s；
2. Valuestate不能跨算子共享；
3. flink1.3 kafka connector不支持partition增长；
4. 与spring整合，出现handler匹配的问题；
5. hadoop的包冲突形成，程序没法正常启动的问题且无异常；

其中比较有意思的是并行度太多，形成barrier对齐花费时间太多的问题。要理解这个问题首先要了解flink在生成checkpoint的过程当中，会在source的插入barrier与正常消息一块儿往下游发射，算子中等到指定的brrier后会触发checkpoint。以下图所：

这是在一个流的状况下，若是有多个流同时进入一个算子处理就会复杂一点。flink在作checkpoint的时候，发现有多个流进入一个算子，先进入这个算子的barrier对应的那段消息就会buffer到算子中等待另外的流对应的barrier也到达才会触发checkpoint，这个buffer再等待的过程称为checkpoint alignment（barrier对齐），以下图：

在线上运行的某个程序的一些算子由于barrier对齐的时间超过50s，形成程序 checkpoint超时失败。对于这个问题，咱们的调优策略是两种，一是尽可能减小并行度，就是让流入一个算子的流尽可能少，若是在4个之内barrier对齐的时间是比较少的。另一种方式，使用at least once的语义替换exactly once的语义，这样checkpoint的时候不会去作barrier对齐，数据到了算子立刻作checkpoint并发送下游。目前 咱们的解决办法是根据不一样的业务场景来区分，若是使用at least once数据保证就能知足业务需求的尽可能用at least once语义。若是不支持的，就减小并行度以此减小barrier对齐的数据量和时间。

平台收益

经过近段时间的平台化建设，在”降本增效“方面的收益主要体如今如下几个方面：

1. 资源利用率提升。目前经过对整个集群的监控，在混合部署的状况下平均cpu利用率在20%左右，在某些cpu密集计算的业务cpu利用率会更高一些；
2. 开发效率提高。好比ETL采集程序的开发，传统开发采集数据、转化、入库大概须要1天左右时间，经过平台化的方式开发简单的ETL程序在1小时内完成开发；
3. 数据处理量大。平均天天处理数据量在80亿条以上；
4. 业务覆盖面广。平台上线业务30+，预计年内突破100+。服务于公司各条业务线，IoT平台，EMS，FMS，智能挂车，企业解决方案，SaaS，硬件部门等。

将来规划

将来对于flink的规划，咱们主要仍是会围绕“降本增效，提供统一的计算平台”为目标，主要聚焦在如下几个方面：
1 .资源隔离更完全。目前的资源隔离使用yarn的默认隔离方式只是对内存隔离，后续须要使用yarn+cgroup对内存和cpu都作隔离。另外会考虑使用yarn的node label作完全机器级别隔离，针对不一样的业务划分不一样类型的机器资源，例如高CPU的任务对应CPU密集型的机器，高IO的任务对应IO比较好的机器；

1. 平台易用性提升。平台包括代码发布、debug、调试、监控、问题排查，一站式解决问题；
2. 减小Code。经过使用Flink SQL+UDF函数的方式，将经常使用的方法和函数进行封装，尽可能用SQL表达业务，提升开发效率。另外也会考虑CEP的模式匹配支持，目前不少业务均可以用动态CEP去支持；
3. 通用的脚手架。在Glink-Framework上持续开发，提供更多的source、sink、工具等，业务封装，简化开发；

此篇文章，摘自于张皓在 「Flink China社区线下 Meetup·成都站」 的技术分享。

更多资讯请访问 Apache Flink 中文社区网站