前言

这几年大数据的飞速发展&＃xff0c;出现了很多热门的开源社区&＃xff0c;其中著名的有 Hadoop、Storm&＃xff0c;以及后来的 Spark&＃xff0c;他们都有着各自专注的应用场景。Spark 掀开了内存计算的先河&＃xff0c;也以内存为赌注&＃xff0c;赢得了内存计算的飞速发展。Spark 的火热或多或少的掩盖了其他分布式计算的系统身影。就像 Flink&＃xff0c;也就在这个时候默默的发展着。

Flink工作原理

Flink是新的stream计算引擎&＃xff0c;用java实现。既可以处理stream data也可以处理batch data&＃xff0c;可以同时兼顾Spark以及Spark streaming的功能&＃xff0c;与Spark不同的是&＃xff0c;Flink本质上只有stream的概念&＃xff0c;batch被认为是special stream。Flink在运行中主要有三个组件组成&＃xff0c;JobClient&＃xff0c;JobManager 和 TaskManager。主要工作原理如下图

用户首先提交Flink程序到JobClient&＃xff0c;经过JobClient的处理、解析、优化提交到JobManager&＃xff0c;最后由TaskManager运行task。

JobClient

JobClient是Flink程序和JobManager交互的桥梁&＃xff0c;主要负责接收程序、解析程序的执行计划、优化程序的执行计划&＃xff0c;然后提交执行计划到JobManager。为了了解Flink的解析过程&＃xff0c;需要简单介绍一下Flink的Operator&＃xff0c;在Flink主要有三类Operator

• Source Operator &＃xff0c;顾名思义这类操作一般是数据来源操作&＃xff0c;比如文件、socket、kafka等&＃xff0c;一般存在于程序的最开始
• Transformation Operator 这类操作主要负责数据转换&＃xff0c;map&＃xff0c;flatMap&＃xff0c;reduce等算子都属于Transformation Operator&＃xff0c;
• Sink Operator&＃xff0c;意思是下沉操作&＃xff0c;这类操作一般是数据落地&＃xff0c;数据存储的过程&＃xff0c;放在Job最后&＃xff0c;比如数据落地到Hdfs、Mysql、Kafka等等。

Flink会将程序中每一个算计解析成Operator&＃xff0c;然后按照算子之间的关系&＃xff0c;将operator组合起来&＃xff0c;形成一个Operator组合成的Graph。如下面的代码解析之后形成的执行计划

DataStream data &＃61; env.addSource(...);data.map(x->new Tuple2(x,1)).keyBy(0).timeWindow(Time.seconds(60)).sum(1).addSink(...)

解析形成执行计划之后&＃xff0c;JobClient的任务还没有完&＃xff0c;还负责执行计划的优化&＃xff0c;这里执行的主要优化是将相邻的Operator融合&＃xff0c;形成OperatorChain&＃xff0c;因为Flink是分布式运行的&＃xff0c;程序中每一个算子&＃xff0c;在实际执行中被分隔为多个SubTask&＃xff0c;数据流在算子之间的流动&＃xff0c;就对应到SubTask之间的数据传递&＃xff0c;SubTask之间进行数据传递模式有两种一种是one-to-one的&＃xff0c;数据不需要重新分布&＃xff0c;也就是数据不需要经过IO&＃xff0c;节点本地就能完成&＃xff0c;比如上图中的source到map&＃xff0c;一种是re-distributed&＃xff0c;数据需要通过shuffle过程重新分区&＃xff0c;需要经过IO&＃xff0c;比如上图中的map到keyBy。显然re-distributed这种模式更加浪费时间&＃xff0c;同时影响整个Job的性能。所以&＃xff0c;Flink为了提高性能&＃xff0c;将one-to-one关系的前后两类subtask&＃xff0c;融合形成一个task。而TaskManager中一个task运行一个独立的线程中&＃xff0c;同一个线程中的SubTask进行数据传递&＃xff0c;不需要经过IO&＃xff0c;不需要经过序列化&＃xff0c;直接发送数据对象到下一个SubTask&＃xff0c;性能得到提升&＃xff0c;除此之外&＃xff0c;subTask的融合可以减少task的数量&＃xff0c;提高taskManager的资源利用率。图1.0中的执行计划&＃xff0c;优化结果如下图&＃xff0c;Flink的subTask融合规则可以参考官方文档。

• 值得注意的是&＃xff0c;并不是每一个SubTask都可以被融合&＃xff0c;对于不能融合的SubTask会独立形成一个Task运行在TaskManager中。
• 改变operator的并行度&＃xff0c;可能会导致不同的优化结果&＃xff0c;同时这也是性能调优的一个重要方式&＃xff0c;例如不显式设置operator的并行度的时候&＃xff0c;默认所有算子的并行度是一样的&＃xff0c;所以会有下图中的优化结果。

我们来分析一下默认情况下可能发生的问题&＃xff0c;假如设置作业的并行度为10&＃xff0c;source明确为kafka&＃xff0c;对应topic只有一个topic&＃xff0c;因为source默认会根据topic的分区数&＃xff0c;决定自己的分区数&＃xff0c;那么10个source subtask只有一个会工作&＃xff0c;而且任务比较重。这样会导致后面的map实际也是有一个subTask在工作&＃xff0c;处理所有的数据&＃xff0c;假如map中的任务比较重&＃xff0c;那么会导致数据倾斜&＃xff0c;性能低下。在source不能改造的情况下&＃xff0c;我们显式减少source的并行度(为了节省资源&＃xff0c;设置1)&＃xff0c;提高map的并行度(增加处理速度&＃xff0c;设为20)。第一眼看上去&＃xff0c;感觉性能提升了不少&＃xff0c;但是在实际情况中却不一定这样。因为调整source和map的并发度&＃xff0c;失去了原有one-to-one数据传递的优势&＃xff0c;导致subTask不能融合&＃xff0c;数据需要reblance&＃xff0c;产生大量的IO&＃xff0c;所以修改并行度也不一定可以提升性能。修改并行度之后&＃xff0c;执行计划的优化结果如下图。所以在实际优化的过程中&＃xff0c;还是要注意结合数据分布和执行计划调优&＃xff0c;理解Flink执行计划的生成过程很有必要。

JobManager

JobManager是一个进程&＃xff0c;主要负责申请资源&＃xff0c;协调以及控制整个job的执行过程&＃xff0c;具体包括&＃xff0c;调度任务、处理checkpoint、容错等等&＃xff0c;在接收到JobClient提交的执行计划之后&＃xff0c;针对收到的执行计划&＃xff0c;继续解析&＃xff0c;因为JobClient只是形成一个operaor层面的执行计划&＃xff0c;所以JobManager继续解析执行计划(根据算子的并发度&＃xff0c;划分task)&＃xff0c;形成一个可以被实际调度的由task组成的拓扑图&＃xff0c;如上图被解析之后形成下图的执行计划&＃xff0c;最后向集群申请资源&＃xff0c;一旦资源就绪&＃xff0c;就调度task到TaskManager。

为了保证高可用&＃xff0c;一般会有多个JobManager进程同时存在&＃xff0c;它们之间也是采用主从模式&＃xff0c;一个进程被选举为Leader&＃xff0c;其他进程为follower。Job运行期间&＃xff0c;只有Leader在工作&＃xff0c;follower在闲置&＃xff0c;一旦Leader挂掉&＃xff0c;随即引发一次选举&＃xff0c;产生新的Leader继续处理Job。JobManager除了调度任务&＃xff0c;另外一个主要工作就是容错&＃xff0c;主要依靠checkpoint进行容错&＃xff0c;checkpoint其实是stream以及executor(TaskManager中的Slot)的快照&＃xff0c;一般将checkpoint保存在可靠的存储中(比如hdfs)&＃xff0c;为了容错Flink会持续建立这类快照。当Flink作业重新启动的时候&＃xff0c;会寻找最新可用的checkpoint来恢复执行状态&＃xff0c;以达到数据不丢失&＃xff0c;不重复&＃xff0c;准确被处理一次的语义。一般情况下&＃xff0c;都不会用到checkpoint&＃xff0c;只有在数据需要积累或处理历史状态的时候&＃xff0c;才需要设定checkpoint&＃xff0c;比如updateStateByKey这个算子&＃xff0c;默认会启用checkpoint&＃xff0c;如果没有配置checkpoint目录的话&＃xff0c;程序会抛异常。

TaskManager

TaskManager是一个进程&＃xff0c;即一个JVM(Flink用java实现)。主要作用是接收并执行JobManager发送的task&＃xff0c;并且与JobManager通信&＃xff0c;反馈任务状态信息&＃xff0c;比如任务分执行中&＃xff0c;执行完等状态&＃xff0c;上文提到的checkpoint的部分信息也是TaskManager反馈给JobManager的。如果说JobManager是master的话&＃xff0c;那么TaskManager就是worker主要用来执行任务。在TaskManager内可以运行多个task。多个task运行在一个JVM内有几个好处&＃xff0c;首先task可以通过多路复用的方式TCP连接&＃xff0c;其次task可以共享节点之间的心跳信息&＃xff0c;减少了网络传输。TaskManager并不是最细粒度的概念&＃xff0c;每个TaskManager像一个容器一样&＃xff0c;包含一个多或多个Slot&＃xff0c;如图1.2。

Slot是TaskManager资源粒度的划分&＃xff0c;每个Slot都有自己独立的内存。所有Slot平均分配TaskManger的内存&＃xff0c;比如TaskManager分配给Solt的内存为8G&＃xff0c;两个Slot&＃xff0c;每个Slot的内存为4G&＃xff0c;四个Slot&＃xff0c;每个Slot的内存为2G&＃xff0c;值得注意的是&＃xff0c;Slot仅划分内存&＃xff0c;不涉及cpu的划分。同时Slot是Flink中的任务执行器(类似Storm中Executor)&＃xff0c;每个Slot可以运行多个task&＃xff0c;而且一个task会以单独的线程来运行。Slot主要的好处有以下几点&＃xff1a;

• 可以起到隔离内存的作用&＃xff0c;防止多个不同job的task竞争内存。
• Slot的个数就代表了一个Flink程序的最高并行度&＃xff0c;简化了性能调优的过程
• 允许多个Task共享Slot&＃xff0c;提升了资源利用率&＃xff0c;举一个实际的例子&＃xff0c;kafka有3个partition&＃xff0c;对应flink的source有3个task&＃xff0c;而keyBy我们设置的并行度为20&＃xff0c;这个时候如果Slot不能共享的话&＃xff0c;需要占用23个Slot&＃xff0c;如果允许共享的话&＃xff0c;那么只需要20个Slot即可(Slot的默认共享规则计算为20个)。

共享Slot&＃xff0c;虽然在flink中允许task共享Slot提升资源利用率&＃xff0c;但是如果一个Slot中容纳过多task反而会造成资源低下(比如极端情况下所有task都分布在一个Slot内)&＃xff0c;在Flink中task需要按照一定规则共享Slot。共享Slot的方式有两种&＃xff0c;SlotShardingGroup和CoLocationGroup&＃xff0c;CoLocationGroup这种方式目前还没有接触过&＃xff0c;如果感兴趣可以查阅官方文档。下面主要介绍一下SlotShardingGroup的用法&＃xff0c;这种共享的基本思路就是给operator分组&＃xff0c;同一组的不同operator的task&＃xff0c;可以共享一个Slot。默认所有的operator属于同一个组“default”&＃xff0c;即所有operator的task可以共享一个Slot&＃xff0c;可以给operator设置不同的group&＃xff0c;防止不合理的共享。Flink在调度task分配Slot的时候有两个重要原则&＃xff1a;

• 同一个job中&＃xff0c;同一个group中不同operator的task可以共享一个Slot
• Flink是按照拓扑顺序从Source依次调度到Sink的

还拿上述的例子来说明Slot共享以及task调度的原理&＃xff0c;如图1.3假设有两个TaskManager(TaskManager1和TaskManager2)&＃xff0c;每个TaskManager有2个Slot(Slot1和Slot2)。为了方便理解Slot共享的流程需要提前定义operator的并发度&＃xff0c;来决定task的调度顺序。假设source/map的并发度为2&＃xff0c;keyBy/window/sink的并发度为4&＃xff0c;那么调度的顺序依次为source/map[1] ->source/map[2] ->keyBy/window/sink[1]->keyBy/window/sink[2]->keyBy/window/sink[3]->keyBy/window/sink[4]。如图1.3为了便于说明流程&＃xff0c;将source/map的并发度设为4&＃xff0c;keyBy/window/sink的并发度设为4。那么首先分配task source/map[1]&＃xff0c;这个时候Slot中还没有task&＃xff0c;分配到TaskManager1中&＃xff0c;然后分配 source/map[2]&＃xff0c;根据Slot共享规则&＃xff0c;source/map[1]和source/map[2] 属于同一operator的不同task&＃xff0c;所以source/map[2]不能共享Slot1&＃xff0c;被分配到TaskManager1的Slot2&＃xff0c;source/map[3]和source/map[4]同样会依次分配不同的Slot&＃xff0c;接下来分配keyBy/window/sink[1],根据Slot共享规则&＃xff0c;它可以和source/map[1]&＃xff0c;共享同一个slot&＃xff0c;所以也被分配到TaskManager1的Slot1中&＃xff0c;keyBy/window/sink的其他任务依次被分配到不同Slot中。图1.4为并行度分别设置为2和4的分配过程&＃xff0c;这里不再展开说明。

总结

上述内容&＃xff0c;主要介绍了&＃xff0c;Flink的基本架构以及Flink执行的基本原理&＃xff0c;重点说明了Flink实现高性能的一些基本原理&＃xff0c;因为写的比较匆忙&＃xff0c;如有错误之处&＃xff0c;欢迎大家评论指正。