Flink原理与实现架构和拓扑概览

  categories:资料  tags:  author:

架构

要了解一个系统,一般都是从架构开始。我们关心的问题是:系统部署成功后各个节点都启动了哪些服务,各个服务之间又是怎么交互和协调的。下方是Flink集群启动后架构图。

当Flink集群启动后,首先会启动一个JobManger和一个或多个的TaskManager。由客户端提交任务给JobManager,JobManager再调度任务到各个TaskManager去执行,然后TaskManager将心跳和统计信息汇报给JobManager.TaskManager之间以流的形式进行数据的传输。上述三者均为独立的JVM进程。

  • 客户为提交作业的客户端,可以是运行在任何机器上(与JobManager环境连通即可)。提交作业后,客户可以结束进程(流的任务),也可以不结束并等待结果返回。
  • JobManager主要负责调度工作并协调任务做检查点,职责上很像Storm的Nimbus。从客户处接收到工作和JAR包等资源后,会生成优化后的执行计划,并以任务的单元调度到各个TaskManager去执行。
  • TaskManager在启动的时候就设置好了槽位数(Slot),每个插槽能启动一个任务,任务为线程。从JobManager处理接收需要部署的任务,部署启动后,与自己的上游建立Netty连接,接收数据并处理。

可以看到Flink的任务调度是多线程模型,并且不同Job / Task混合在一个TaskManager进程中。虽然这种方式可以有效提高CPU利用率,但是个人不太喜欢这种设计,因为不仅缺少资源隔离机制,同时也不方便调试。类似Storm的进程模型,一个JVM中只跑该Jobs Tasks实际应用中更为合理。

工作例子

本文所示例子为flink-1.0.x版本

我们使用Flink自带的例子包中的SocketTextStreamWordCount,这是一个从socket流中统计单词出现次数的例子。

首先,使用netcat的启动本地服务器:

$ nc -l 9000

然后提交Flink程序

$ bin / flink运行示例/ streaming / SocketTextStreamWordCount.jar \
  --hostname 10.218.130.9 \
  -  9000

在netcat端输入单词并监控taskmanager的输出可以看到单词统计的结果。

SocketTextStreamWordCount 的具体代码如下:

public  static  void  main (String [] args) throws Exception {
  //检查输入
  最终 ParameterTool PARAMS = ParameterTool.fromArgs(参数);
  ...
  //设置执行环境
  最终 StreamExecutionEnvironment ENV = StreamExecutionEnvironment.getExecutionEnvironment();
  //获取输入数据
  DataStream <String> text =
      env.socketTextStream(params.get(“hostname”),params.getInt(“port”),'\ n',0);
  DataStream <Tuple2 <String,Integer >>计数=
      //分开成对的行(2元组),包含:(word,1)
      text.flatMap(new Tokenizer())
          //由元组字段“0”组合,并将元组字段“1”
          .keyBy(0)
          .sum(1);
  counts.print();
  
  //执行程序
  env.execute(“SocketTextStream示例中的WordCount”);
}

我们将最后一行代码替换env.execute成System.out.println(env.getExecutionPlan());并并在本地运行该代码(并发度设为2),可以得到该拓扑的逻辑执行计划图的JSON串,将JSON串粘贴到http://flink.apache.org/可视化/中,能可视化该执行图。

但是并不是最终在Flink中运行的执行图,只是一个表示拓扑节点关系的计划图,在Flink中对应了SteramGraph。另外,提交拓扑后(并发度设为2)还能在UI中看到另一张执行计划图,如下所示,该图对应了Flink中的JobGraph。

图形

看起来有点乱,怎么有这么多不一样的图。实际上,还有更多的图.Flink中的执行图可以分成四层:StreamGraph – > JobGraph – > ExecutionGraph – >物理执行图。

  • StreamGraph:是根据用户通过Stream API编写的代码生成的最初的图。用来表示程序的拓扑结构。
  • JobGraph: StreamGraph经过优化后生成了JobGraph,提交给JobManager的数据结构。主要的优化为,将多个符合条件的节点链在一起作为一个节点,这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。
  • ExecutionGraph: JobManager根据JobGraph生成ExecutionGraph.ExecutionGraph是JobGraph的并行化版本,是调度层最核心的数据结构。
  • 物理执行图: JobManager根据ExecutionGraph对工作进行调度后,在各个TaskManager上部署任务后形成的“图”,并不是一个具体的数据结构。

例如上文中的2个并发度(来源为1个并发度)的SocketTextStreamWordCount四层执行图产品的演变过程如下图产品所示(点击查看大图):

这里对一些名词进行简单的解释。

  • StreamGraph:根据用户通过Stream API 编写的代码生成的最初的图。
    • StreamNode:用来代表运算符的类,并具有所有相关的属性,如并发度,入边和出边等。
    • StreamEdge:表示连接两个StreamNode的边。
  • JobGraph: StreamGraph经过优化后生成了JobGraph,提交给JobManager的数据结构。
    • JobVertex:经过优化后符合条件的多个StreamNode可能会连锁在一起生成一个JobVertex,即一个JobVertex包含一个或多个运营商,JobVertex的输入是JobEdge,输出是IntermediateDataSet。
    • IntermediateDataSet:表示JobVertex的输出,即经过运营商处理产生的数据集.producer是JobVertex,消费者是JobEdge。
    • JobEdge:代表了工作图中的一条数据传输通道.source是IntermediateDataSet,目标是JobVertex。即数据通过JobEdge由IntermediateDataSet传递给目标JobVertex。
  • ExecutionGraph: JobManager根据JobGraph生成ExecutionGraph.ExecutionGraph是JobGraph的并行化版本,是调度层最核心的数据结构。
    • ExecutionJobVertex:和JobGraph中的JobVertex一一对应。每一个ExecutionJobVertex都有和并发度一样多的ExecutionVertex。
    • ExecutionVertex:表示ExecutionJobVertex的其中一个并发子任务,输入是ExecutionEdge,输出是IntermediateResultPartition。
    • IntermediateResult:和JobGraph中的IntermediateDataSet一一对应每一个IntermediateResult有与下游ExecutionJobVertex相同并发数的IntermediateResultPartition。
    • IntermediateResultPartition:表示ExecutionVertex的一个输出分区,制片人是ExecutionVertex,消费者是若干个ExecutionEdge。
    • ExecutionEdge:表示ExecutionVertex的输入,源是IntermediateResultPartition,目标是ExecutionVertex.source和目标都只能是一个。
    • 执行:执行一个ExecutionVertex的一次尝试。当发生故障或者数据需要重算的情况下ExecutionVertex可能会有多个ExecutionAttemptID。一个执行通过ExecutionAttemptID来唯一标识.JM和TM之间关于任务的部署和任务状态更新都是通过ExecutionAttemptID来确定消息接受者。
  • 物理执行图: JobManager根据ExecutionGraph对工作进行调度后,在各个TaskManager上部署任务后形成的“图”,并不是一个具体的数据结构。
    • 任务:执行被调度后在分配的TaskManager中启动对应的Task.Task包裹了具有用户执行逻辑的运算符。
    • ResultPartition:代表由一个任务的生成的数据,和ExecutionGraph中的IntermediateResultPartition一一对应。
    • ResultSubpartition:是ResultPartition的一个子分区。每个ResultPartition包含多个ResultSubpartition,其数目要由下游消费任务数和DistributionPattern来决定。
    • InputGate:代表任务的输入封装,和JobGraph中JobEdge一一对应每个InputGate消费了一个或多个的ResultPartition。
    • InputChannel:每个InputGate会包含一个以上的InputChannel,和ExecutionGraph中的ExecutionEdge一一对应,也和ResultSubpartition一对一地相连,即一个InputChannel接收一个ResultSubpartition的输出。

那么Flink为什么要设计这4张图呢,其目的是什么呢?Spark中也有多张图,数据依赖图以及物理执行的DAG。其目的都是一样的,就是解耦,每张图各司其职位,每张图对应了工作不同的阶段,更方便做该阶段的事情。我们给出更完整的Flink Graph的层次图。

首先我们看到,JobGraph之上除了StreamGraph还有OptimizedPlan.OptimizedPlan是由Batch API转换而来的.StreamGraph是由Stream API转换而来的。为什么API不直接转换成JobGraph?因为,Batch和Stream的图结构和优化方法有很大的区别,比如批量有很多执行前的预分析用来优化图的执行,而这种优化并不适合流,所以通过OptimizedPlan来做批量的优化会更方便和清晰,也不会影响Stream.JobGraph的责任就是统一Batch和Stream的图,用来描述清楚一个拓扑图的结构,并且做了chaining的优化,chaining是普适于Batch和Stream的,所以在这一层做掉.ExecutionGraph的责任是方便调度和各个任务状态的监控和跟踪,所以ExecutionGraph是并行化的JobGraph。而“物理执行图”就是最终分布在各个机器上运行着的任务了。所以可以看到,这种解耦方式极大地方便了我们在各个 所做的工作,各个层之间是相互隔离的。

后续的文章,将会详细介绍Flink是如何生成这些执行图的。由于我目前关注Flink的流程处理功能,所以主要有以下内容:

如何生成StreamGraph

  1. 如何生成JobGraph
  2. 如何生成ExecutionGraph
  3. 如何进行调度(如何生成物理执行图)

 

来源: https://zhuanlan.zhihu.com/p/27576821



快乐成长 每天进步一点点      京ICP备18032580号-1