flink图流转之StreamGraph

发表于 | 分类于 | | |
字数统计 1.6k | 阅读时长 7

flink DAG图流转分析

前言

每次我们编写完flink作业，跑任务的时候都会在flink-ui上展示一个作业的DAG图，那么这个图是如何形成的呢？本文就和你一起来揭开flink执行图生成的神秘面纱~

##总览
在flink中的执行图可以分为4层StreamGraph -> JobGraph -> ExecutionGraph -> 物理执行图。

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

今天我们就来看下streamGraph的生成

StreamGraph的生成

组件

1
2
3
StreamGraph：根据用户通过 Stream API 编写的代码生成的最初的图。
StreamNode：用来代表 operator 的类，并具有所有相关的属性，如并发度、入边和出边等。
StreamEdge：表示连接两个StreamNode的边。

flink任务从定义一个运行环境开始streamExecutionEnvironment，流计算任务起始于addSource,我们来看这个函数，addSource之后生成了一个DataStream. DataStream的构造函数参数接收一个StreamTransformation类型的对象，这个对象反映了流之间的转换操作。但是这个transformation和operation不是一一对应的。一些分区操作：union，split/select，partition只是逻辑概念，并不会在最后的dag图上显示出来。
在生成datastream之后，经历DataStream.java中定义的一些api算子，完成业务逻辑的定义，在这之中可能包含以下的转化：

假设一个场景：

1
2
addsource -> map -> filter -> connect -> flatmap 
-> keyby -> window -> apply -> addSink -> excute
  1. addSource创建生成一个SingleOutputStreamOperator 本质上是一个带有transformation=”SourceTransformation”的datastream
  2. map创建生成一个OneInputTransformation 并调用getExecutionEnvironment().addOperator(resultTransform)将其添加入env的List<StreamTransformation<?>>中
  3. filter 通过 map相同操作
  4. connect 直接返回一个ConnectedStreams不是Datastream的子类
  5. flatmap 生成一个TwoInputTransformation将其添加入env的List<StreamTransformation<?>>中，并返回一个SingleOutputStreamOperator,并且返回的Datastream中包含的是当前这个transformation
  6. keyby 生成一个keyedStream，这里直接生成一个PartitionTransformation 替代了父类DataStream中的transformation
  7. window 生成windowStream
  8. apply调用将生成一个OneInputTransformation,增加至List<StreamTransformation<?>>
  9. addSink 获取 SinkTransformation

其中每一次创建OneInputTransformation都是基于Datastream的当前的transformation来创建的，也就是说keyby之后的PartitionTransformation信息也加入了.

1
2
3
4
5
6
new OneInputTransformation<>(
			this.transformation,
			operatorName,
			operator,
			outTypeInfo,
			environment.getParallelism());

好了到这里已经获取了各个流程的streamtransformation,最后调用execute方法，截取了流式环境下的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public JobExecutionResult execute(String jobName) throws Exception {
	Preconditions.checkNotNull("Streaming Job name should not be null.");

	StreamGraph streamGraph = this.getStreamGraph();
	streamGraph.setJobName(jobName);

	transformations.clear();

	// execute the programs
	if (ctx instanceof DetachedEnvironment) {
		LOG.warn("Job was executed in detached mode, the results will be available on completion.");
		((DetachedEnvironment) ctx).setDetachedPlan(streamGraph);
		return DetachedEnvironment.DetachedJobExecutionResult.INSTANCE;
	} else {
		return ctx
			.getClient()
			.run(streamGraph, ctx.getJars(), ctx.getClasspaths(), ctx.getUserCodeClassLoader(), ctx.getSavepointRestoreSettings())
			.getJobExecutionResult();
	}
}

其实主要调用的就是

1
StreamGraphGenerator.generate(this, transformations);

每一个OneInputTransformation都会记录他的上游的input的transformation，在StreamGraphGenerator.generate主要针对不同的transformation进行不同的转化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
private Collection<Integer> transform(StreamTransformation<?> transform) {

		if (alreadyTransformed.containsKey(transform)) {
			return alreadyTransformed.get(transform);
		}

		LOG.debug("Transforming " + transform);

		if (transform.getMaxParallelism() <= 0) {

			// if the max parallelism hasn't been set, then first use the job wide max parallelism
			// from theExecutionConfig.
			int globalMaxParallelismFromConfig = env.getConfig().getMaxParallelism();
			if (globalMaxParallelismFromConfig > 0) {
				transform.setMaxParallelism(globalMaxParallelismFromConfig);
			}
		}

		// call at least once to trigger exceptions about MissingTypeInfo
		transform.getOutputType();

		Collection<Integer> transformedIds;
		if (transform instanceof OneInputTransformation<?, ?>) {
			transformedIds = transformOneInputTransform((OneInputTransformation<?, ?>) transform);
		} else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
			transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
		} else if (transform instanceof SourceTransformation<?>) {
			transformedIds = transformSource((SourceTransformation<?>) transform);
		} else if (transform instanceof SinkTransformation<?>) {
			transformedIds = transformSink((SinkTransformation<?>) transform);
		} else if (transform instanceof UnionTransformation<?>) {
			transformedIds = transformUnion((UnionTransformation<?>) transform);
		} else if (transform instanceof SplitTransformation<?>) {
			transformedIds = transformSplit((SplitTransformation<?>) transform);
		} else if (transform instanceof SelectTransformation<?>) {
			transformedIds = transformSelect((SelectTransformation<?>) transform);
		} else if (transform instanceof FeedbackTransformation<?>) {
			transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
		} else if (transform instanceof CoFeedbackTransformation<?>) {
			transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
		} else if (transform instanceof PartitionTransformation<?>) {
			transformedIds = transformPartition((PartitionTransformation<?>) transform);
		} else if (transform instanceof SideOutputTransformation<?>) {
			transformedIds = transformSideOutput((SideOutputTransformation<?>) transform);
		} else {
			throw new IllegalStateException("Unknown transformation: " + transform);
		}

		// need this check because the iterate transformation adds itself before
		// transforming the feedback edges
		if (!alreadyTransformed.containsKey(transform)) {
			alreadyTransformed.put(transform, transformedIds);
		}

		if (transform.getBufferTimeout() > 0) {
			streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());
		}
		if (transform.getUid() != null) {
			streamGraph.setTransformationUID(transform.getId(), transform.getUid());
		}
		if (transform.getUserProvidedNodeHash() != null) {
			streamGraph.setTransformationUserHash(transform.getId(), transform.getUserProvidedNodeHash());
		}

		if (transform.getMinResources() != null && transform.getPreferredResources() != null) {
			streamGraph.setResources(transform.getId(), transform.getMinResources(), transform.getPreferredResources());
		}

		return transformedIds;
	}

可以看到他里面的方法都是递归调用transform(input)方法，然后通过alreadyTransformed数据结构，避免重复计算，所以我们最终看的时候最先是从source处进行的，也就是从上游到下游进行转化

  1. 如果已经在alreadyTransformed数据结构中那么就直接返回transformation的id
  2. 分别有addSource，addOperator，addSink，addCoOperator，addEdge的不同操作来生成streamGraph中的不同节点
  3. addEdge建立每一个transformation和他所有上游输入节点的连线

在streamgraph中还建立了几个虚拟的节点，这几个节点主要针对的是partition，split/select，sideoutput的操作。

1
2
3
private Map<Integer, Tuple2<Integer, List<String>>> virtualSelectNodes;
private Map<Integer, Tuple2<Integer, OutputTag>> virtualSideOutputNodes;
private Map<Integer, Tuple2<Integer, StreamPartitioner<?>>> virtualPartitionNodes;

在进行这些操作时，会添加一个唯一的虚拟节点

1
2
3
4
5
6
//记录了上游某个transformId到下游的partition方式
virtualPartitionNodes.put(virtualId, new Tuple2<Integer, StreamPartitioner<?>>(originalId, partitioner));  
//记录上游的不同outputTag，用以将部分数据从该tag输出
virtualSideOutputNodes.put(virtualId, new Tuple2<>(originalId, outputTag));
//记录一个上游的select虚拟节点
virtualSelectNodes.put(virtualId, new Tuple2<Integer, List<String>>(originalId, selectedNames));

经过一些列的addNode以及addEdge之后，streamGraph已经生成。关于其他几个graph的生成请听下回的分解

谢谢支持