基于Flink实时计算的自动化流控制算法
2020-08-12樊春美朱建生单杏花杨立鹏
樊春美，朱建生，单杏花，杨立鹏，李 雯
(中国铁道科学研究院，北京 100081)
0 引 言
随着铁路售票系统业务的多样化，来自各个业务系统的日志解析变得越来越复杂。对于日志数据进行实时的采集和解析是当前需要解决的重要问题。鉴于大数据的应用比较广泛，尤其是分布式的架构和实时计算框架，如Flink[1]、Spark[2]计算平台变得越来越热门，因此现在大部分互联网公司都会采用实时计算来完成日志的采集和存储。而对于复杂的铁路售票系统，不仅需要针对不同服务层进行日志的采集，还需要对不同业务场景的不同类型的日志进行解析，为了业务数据分析的需要，可能同一份数据还要存储到不同的渠道中；与此同时系统会随着生产的需求而不断完善，因此也会导致日志存储格式的变动，从而需要不停地调整解析程序来适应新的需求。在每一次变动中，都需要测试整套解析程序，开发效率较慢，甚至对于一个数据流的解析要启动多个程序，造成资源的浪费，维护多个程序也变得更加复杂。于是可以采用现在常用的改变配置文件的方式来实时更新解析相关的配置。但是分布式的计算框架刷新配置文件存在着两个问题，一个是配置文件的存放问题，一般分布式的计算框架，真正的计算程序是在每个节点上执行的，配置文件如果在master节点上，而每个执行节点是不能读取master节点的配置文件的，如果每个执行节点都放一份配置文件，也无法保证每个执行节点的路径是一致的，只要其中的一个节点路径不一致就会导致程序出错，无法执行；另外一个问题是，假设能够保证每台机器的路径都是一致的，配置文件可以放在每台机器上，这时程序每解析一条数据都要读取一次配置文件，频繁的读取必然会影响解析的效率。由于对于分布式的计算框架刷新配置存在着上述两个问题，文中提出了一种通过自动化流控制的方法，实现配置文件的实时刷新。
1 相关工作
目前针对大型数据的分析框架主要有Hadoop、Spark、Flink等[3-5]。以Hadoop为代表的大数据技术的出现，可以很好地解决大量静态数据集的数据处理与分析，但是很多数据都是实时产生的，用户希望可以实时地处理这些数据，这就需要使用流计算技术来实时处理这些数据，及时产出应用价值。
而Apache Spark是专为大规模数据处理而设计的快速通用的计算引擎，是一种与Hadoop相似的开源集群计算环境，其拥有Hadoop MapReduce所具有的优点，但与MapReduce存在的不同是任务中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark除了能够提供交互式查询外，还可以优化迭代工作，能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。
随着Spark计算引擎的出现，促进了上层应用快速发展，如对各种迭代计算的性能以及对流计算和SQL等的支持，Flink就在这样的背景下诞生了。Flink的突出优势是对流计算的支持及更一步的实时性。为了对流式引擎处理有更深刻的认识，Sanket Chintapalli等人[6]基于常用的流式计算构建了三种不同的流引擎，进行对比分析。其中Spark streaming是将流划分解成一系列短小的批处理作业，Flink才是真正的流处理，批处理只是流数据的一个极限特例而已。Flink不仅可以支持本地的快速迭代，以及一些环形的迭代任务，而且Flink可以定制化内存管理，其并没有将内存完全交给应用层，因此Flink处理大数据速度快，能够更好地满足大数据背景下应用实时计算平台的需求。
由于Flink框架的优势，现在有很多关于Flink应用的相关研究。蔡鲲鹏[7]研究了Flink的概念、生态系统和相关技术等理论基础并对Hadoop和Flink在处理大批量数据上的耗时和准确率进行了对比分析，针对不同的流式处理平台，分析总结了Flink所面临的一些挑战，为Flink的进一步研究提供了参考。麦冠华等[8]基于Flink的计算框架，设计了对大规模轨迹数据进行实时运动模式检测的算法，弥补了对于当前大规模轨迹数据只能做范围查询、近邻查询的简单处理的不足，很好地应用了Flink实时计算的优势。Marciani G[9]利用Flink框架对社交网络进行实时分析，系统架构的设计重点在于利用Flink的并行性和内存效率，以便能够在分布式基础设施上有效地处理大容量数据流。不仅Flink的应用比较广泛，对其优化的相关研究也比较多。Verbitskiy I等[10]分析了Flink的执行效率，通过各种实验评估表明Apache Flink的性能是高度依赖于问题的；李梓杨等人[11]通过针对大数据流式计算平台中输入数据流急剧上升所导致的计算延迟升高问题进行优化，有效地提高了现在Flink框架集群的吞吐量；文献[12-13]对基于多查询和状态管理进行了优化，研究了Flink的可扩展性等等。但是对于Flink使用过程中数据解析逻辑的控制研究相对较少。
文中主要从对Flink进行实时解析时逻辑的更改角度进行优化，通过使用流控制的方式，减少代码的开发量，提高Flink应用实时解析的效率。
2 实时计算架构
随着业务越来越复杂，需要采集和存储的数据越来越多，由于存在着不同的业务系统，日志的存储格式多种多样。为了对不同的日志进行解析，同时能够根据不同的需求将解析的数据输出到相应的存储空间，需要开发一套满足灵活地适配各种日志格式的数据解析架构，从而减少同类解析代码的开发，将不同的数据解析进行集中式的管理。文中基于队列和分布式流处理架构构建了大数据的实时采集计算和存储平台。数据处理架构如图1所示。
1.数据采集。
日志数据都是实时产生的，在采集的过程中，也是在不断生成的，因此数据采集模块需要完成实时采集。目前应用较多的有Tcollector、Filebeat[14]等采集工具。
其中Filebeat具有两个较大的优势：
(1)性能稳健。
图1 数据处理架构
无论什么样的应用都可能存在程序中断的情况，Filebeat能够读取并转发日志行，如果出现中断，还会在一切恢复正常后，从中断前停止的位置继续开始。
(2)部署简单。
Filebeat内置有多种模块(Apache、System、MySQL等等)，可以针对常见格式的日志大大简化收集、解析和可视化过程。
基于Filebeat的优势，在构建数据的采集平台时采用该服务进行日志的实时采集。在部署采集程序时，将不同的业务发送到不同的topic数据流中；通过Filebeat的配置文件实现数据采集的机器、日志文件、采集的路径、数据的输出端的配置。
2.数据传输。
数据传输采用Kafka[15]队列，每个topic队列作为一个单独的数据流，并与数据的采集和解析构成完整的数据处理流。除了采集业务数据的数据流，这里增加一个空流，用来进行流控制。Kafka采集的各个数据流如图2所示。
图2 采集的数据流
3.数据解析。
数据解析部分需要获取多种数据的配置，如系统配置、数据源配置、数据解析逻辑配置、数据存储配置、监控配置等。数据解析模块主要是通过使用Flink的各种算子组合完成业务数据解析逻辑的。该模块是整个实时采集计算和存储平台的核心部分，对数据的实时计算能力要求较高。该模块不仅需要完成对数据流的实时解析，同时还要支持对数据流解析的实时更改。例如一个流能够解析多个topic的数据，一个topic能够通过解析程序分流到不同的存储路径，对一个流能够实时地更改解析逻辑，而不需要重启。
4.数据存储。
日志数据不仅用来进行业务的分析，还需要对各种业务的指标进行监控，所以同一份日志的数据需要存储到不同的存储介质，因此数据流的输出结果也会有多种，如hdfs、hive、clickhouse、opentsdb等多个存储渠道。
3 自动化流控制算法
基于第2节介绍的实时计算架构，提出了使用更新算子的方式来改变数据流的解析逻辑。
3.1 流迭代算法
现在通用的数据流处理方式是流stream1处理完，将得到的结果作为stream2的输入，在stream2流的处理中完成对stream1的结果的处理，将stream流的解析通过不同的map逻辑依次处理，直到得到想要的输出结果。解析逻辑如图3所示。
图3 解析逻辑
而第2节介绍的实时计算架构，不再使用这种方式进行业务逻辑的处理，而是通过配置文件来实现，为了满足通过更新配置文件代替代码开发完成日志解析的需求，提出流迭代的算法，具体算法逻辑如下：
输入：需要解析的数据流stream。
输出：解析结果。
Step1：将每个要处理的数据流的名称通过hashmap进行存储，假设
Step2：按照对datastream1的流处理算子得到流处理结果dataset1;
Step3：更新hashmap中stream1的value值为dataset1；
Step4：遍历下一个需要处理的算子，直接读取key=stream1的value值，对stream1的value值执行相应的解析逻辑得到数据集dataset2；
Step5：更新stream1的value值为dataset2；
Step6：依次迭代对数据流处理的各个算子，直到完成所有的解析逻辑，最后结果依然保存在stream1中。
通过流迭代的方法，每次算子执行的时候都是对同一个stream1进行处理，只需要遍历定义好的算子即可，这样很多算子在不同的数据流解析中可以共用，不仅可以减少代码的开发量，还可以把开发的重点放在业务逻辑处理中，解析日志的程序开发变得更加简单。在算子中还可以添加复制的算子，将一个数据流复制成多个数据流，再针对不同的数据流配置不同的日志解析算子，实现分流的效果。
3.2 流控制算法
程序的重新启动会中断正在运行的解析逻辑，有些数据实时性要求较高，中间重启程序会造成一些数据的缺失。同时针对3.1介绍的流切换算法中的stream值更新的时候也需要实时地传入，因此文中设计了免更新、免重启的流控制算法。
输入：解析算子γ，算子γ是可以实现数据流选择和各种解析业务逻辑的配置，里面通过设置一个参数source，实现对不同数据流的解析逻辑控制。
输出：按照算子指定的执行逻辑输出结果。
Step1：假设需要解析的数据流为dataA,在现有需要解析的数据流中增加一个空的数据流temp，该数据流开始时不存储任何数据，同时增加一个内部类的变量用来存储解析的算子γ；
Step2：在实时的代码解析逻辑中，增加一个对temp流的解析；
Step3：在需要更新解析逻辑时，通过注入的方式将最新的解析逻辑注入到temp流中；
Step4：通过解析temp流中的数据，获取针对当前数据流的解析逻辑，并更新为γ的值；
Step5：再次解析数据流dataA的时候，就会使用最新的解析逻辑来处理数据，从而实现解析逻辑的实时控制。
算法的实现逻辑如图4所示，数据流处理主要分为数据的采集和解析，业务数据流主要是从各个业务系统实时采集对应的数据，而逻辑数据流是在需要解析某个业务数据时，传入业务流对应的解析逻辑；在数据解析环节首先获取解析逻辑的解析算子，从而实现对业务数据流解析的控制。
图4 业务数据流处理流程
3.3 temp流中的解析算子γ
注入temp数据流的解析算子γ配置如下：
#系统配置
#数据源
#业务逻辑(任务解析)
#配置输出
#监控配置
通过java生成相应的文件，注入到temp数据流中。在整个架构中有一个控制类，该类通过读取XML文件，解析一个配置类，配置的成员包括系统配置类、数据源类、业务逻辑算子类、配置输出sink类等，其中算子类在实现的时候会继承一个基类，这样不同类型的算子都可以组成一个基类算子的list列表。
4 算法应用实例
以从Kafka队列接收数据，使用Flink解析，输出到hdfs、Opentsdb存储数据中为例，介绍该算法在具体的实时流计算框架中的应用。
4.1 配置文件实例
下面给出了注入temp的主要配置信息。
4.2 Flink代码解析
Flink是一种典型的分布式计算，对于内部类外的变量会在程序启动后存储在master节点上，且后续都不能改变，而对于内部类中的变量在每次代码执行时都会执行，因此采用3.2介绍的流控制方法在内部类中添加一个变量，用来存储解析算子，在对不同业务数据进行处理时，更新这个值就可以达到解析的目的。
伪代码逻辑如下：
创建一个临时变量str=temp;
#解析数据流DataA
{
#根据str，实现map的解析
DataStream
}
#解析数据流temp
{
Parse(temp_map)
输入新的解析算子new_temp
If(Source=stream1){ str=new_temp;}
else{Return;}
4.3 实验对比
目前生产上共配置了5台CPU16核，内存为64 G的服务器，搭建了实时解析架构平台，每秒的日志处理流量大概15 W左右，处理业务日志种类多达50个，随着业务的变动，实时平台的调整也会比较频繁。
完成一个在线运行的业务更改逻辑过程对比如下：
不使用文中提出的自动化流控制算法的业务处理过程：
(1)从Filebeat采集的每条数据中获取需要的字段；
(2)对指定的日志数据写代码进行解析，调试；
(3)查看解析结果，是否满足业务的需求；
(4)写入库的sink代码，调试；
(5)查看入库的结果；
(6)将代码打包、上传jar包到各台服务上；
(7)重启各个进程，查看程序是否正常启动。
使用文中提出的自动化流控制算法的业务处理过程：
(1)在配置文件中实现所有的逻辑；
(2)查看解析和入库的结果；
(3)将配置文件注入到topic流中；
(4)自动生效，完成逻辑的更改。
消耗和用时结果对比如表1所示。
表1 是否使用流处理的对比结果
从实现的流程可以看到，使用文中提出的自动化流控制算法可以很大程度地减少代码的开发与测试，减少生产部署的工作量，最大程度地保证了生产数据的实时性。
5 结束语
通过Filebeat、Kafka队列和Flink流式处理架构，构建了一套实时数据流解析的平台，不仅能够针对大量的数据进行实时解析，还能够满足实时更新解析逻辑的策略，极大地提高了生产数据的采集和分析效率。通过该实时解析架构平台，可以实现多流合并，单流分流，及业务逻辑实时更新等多个功能，提高了分布式流处理架构Flink的应用性能，为当前各个互联网公司复杂业务逻辑的大数据处理提供了解决方案，具有一定的现实意义。后续将会对Flink使用的资源做进一步的优化，提高数据解析平台的资源利用率。
