Flink 基本工作原理

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

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

JobClient

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

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

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

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

JobManager

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

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

TaskManager

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

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

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

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

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

   

总结

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

参考资料

• https://ci.apache.org/projects/flink/flink-docs-master/internals/job_scheduling.html?spm=a2c4e.11153940.blogcont64819.14.4cc928ce5F2w98
• https://ci.apache.org/projects/flink/flink-docs-master/concepts/programming-model.html
• https://yq.aliyun.com/articles/64819
• https://blog.csdn.net/lisi1129/article/details/54844919
• Learning Apache Flink

今天的文章Flink 基本工作原理分享到此就结束了，感谢您的阅读，如果确实帮到您，您可以动动手指转发给其他人。