Job的提交流程（一）
一个Flink作业，从client提交到真正的执行，其 Graph 的转换会经过下面三个阶段（第四个阶段是作业真正执行时的状态，都是以 task 的形式在 TM 中运行）：
StreamGraph：根据编写的代码生成最初的 Graph，它表示最初的拓扑结构；
JobGraph：这里会对前面生成的 Graph，做一些优化操作（比如: operator chain 等），最后会提交给 JobManager；
ExecutionGraph：JobManager 根据 JobGraph 生成 ExecutionGraph，是 Flink 调度时依赖的核心数据结构；
物理执行图：JobManager 根据生成的 ExecutionGraph 对 Job 进行调度后，在各个 TM 上部署 Task 后形成的一张虚拟图。
首先要熟悉以下概念：DataStream、Transformation、StreamOperator、Function
DataStream
image.png
DateStream实际上就是对相同类型的数据流操作进行的封装，主要是通过Transformations将数据流转换成另一个流，常用的api:
keyBy()、join()、union()、map()、filter()、flatMap()等。
Transformation
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// get input data by connecting to the socket
DataStream<String> text = env.socketTextStream(hostname, port, "\n");
// parse the data, group it, window it, and aggregate the counts
DataStream<WordWithCount> windowCounts =
text.flatMap(
new FlatMapFunction<String, WordWithCount>() {
@Override
public void flatMap(
String value, Collector<WordWithCount> out) {
for (String word : value.split("\\s")) {
out.collect(new WordWithCount(word, 1L));
}
}
})
.keyBy(value -> value.word)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.reduce(
new ReduceFunction<WordWithCount>() {
@Override
public WordWithCount reduce(WordWithCount a, WordWithCount b) {
return new WordWithCount(a.word, a.count + b.count);
}
});
// print the results with a single thread, rather than in parallel
windowCounts.print().setParallelism(1);
env.execute("Socket Window WordCount");
首先看env.socketTextStream()，调用了StreamExecutionEnvironment的addSource()方法，会讲SourceFunction封装到StreamSource中，之后会讲StreamSource封装到SourceTransformation中。所以最终执行业务核心逻辑的是各种Function，经过各种转换会生成对应的Transformation。
private <OUT> DataStreamSource<OUT> addSource(
final SourceFunction<OUT> function,
final String sourceName,
@Nullable final TypeInformation<OUT> typeInfo,
final Boundedness boundedness) {
checkNotNull(function);
checkNotNull(sourceName);
checkNotNull(boundedness);
TypeInformation<OUT> resolvedTypeInfo =
getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);
boolean isParallel = function instanceof ParallelSourceFunction;
clean(function);
final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);
//DataStreamSource 其实是SingleOutputStreamOperator的子类
return new DataStreamSource<>(
this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
}
再看flatMap()操作，其实和上述类似，会将Function封装成StreamOperator，再封装成Transformation，最终都返回拥有当前算子和环境的DataStream对象，OneInputTransformation算法拥有当前算子的上游算子对象。整个封装过程Function->StreamOperator->Transformation
image.png
public <R> SingleOutputStreamOperator<R> flatMap(
FlatMapFunction<T, R> flatMapper, TypeInformation<R> outputType) {
return transform("Flat Map", outputType, new StreamFlatMap<>(clean(flatMapper)));
}
protected <R> SingleOutputStreamOperator<R> doTransform(
String operatorName,
TypeInformation<R> outTypeInfo,
StreamOperatorFactory<R> operatorFactory) {
// read the output type of the input Transform to coax out errors about MissingTypeInfo
transformation.getOutputType();
OneInputTransformation<T, R> resultTransform =
new OneInputTransformation<>(
/**
* han_pf
* 记录当前transformation的输入transformation
*/
this.transformation,
operatorName,
operatorFactory,
outTypeInfo,
environment.getParallelism());
@SuppressWarnings({"unchecked", "rawtypes"})
SingleOutputStreamOperator<R> returnStream =
new SingleOutputStreamOperator(environment, resultTransform);
/**
* han_pf
* 存储所有的stransformation到env中。
*/
getExecutionEnvironment().addOperator(resultTransform);
return returnStream;
}
StreamOperator
Operator 基类的是 StreamOperator，它表示的是对 Stream 的一个 operation，它主要的实现类如下：
AbstractUdfStreamOperator：会封装一个 Function，真正的操作是在 Function 中的实现。
OneInputStreamOperator：如果这个 Operator 只有一个输入，实现这个接口即可，processElement() 方法需要自己去实现，主要做业务逻辑的处理；
TwoInputStreamOperator：如果这个 Operator 是一个二元操作符，是对两个流的处理，比如：双流 join，那么实现这个接口即可，自己去实现 processElement1() 和 processElement2() 方法。
image.png
Function
Function 是 Transformation 最底层的封装，用户真正的处理逻辑是在这个里面实现的，包括前面示例中实现的 FlatMapFunction 对象等。
image.png
以上熟悉各个概念之后，以及各个类之间的关系，对于后边StreamGraph的生成过程有极大的帮助。
