Flink-DataStream流处理应用(Local模式下)运行流程-源码分析

Posted 2020-12-17 mengyao

 

Flink的部署模式，有Local、Cluster和Cloud模式，本案例，在Local模式下分析，Flink的DataStream流处理应用程序的运行流程

1、自定义DataStream的应用程序，执行execute方法

1.1、初始化StreamExecutionEnvironment，在local模式下，调用的是1939行，即无参数的createLocalEnvironment()

 

 

1.2、查看createLocalEnvironment方法的定义，内部调用的是createLocalEnvironment(defaultLocalParallelism)，传入了一个int类型的默认并行度参数

注意默认并行度参数的值是这么生成的：

 1.3、查看createLocalEnvironment(defaultLocalParallelism)方法的定义，内部调用的是createLocalEnvironment(defaultLocalParallelism，configuration)，传入了两个参数，分别是并行度参数和配置对象

 

1.4、我们看到了，原来在IDE中，用的是LocalStreamEnvironment对象

 

 

1.5、在new LocalStreamEnvironment构造方法中，通过vlidateAndGetConfiguration方法设置了2个配置：execution.target=local 和 execution.attached=true；然后设置并行度为1

1.6、在new LocalStreamEnvironment的时候，也调用了父类StreamExecutionEnvironment(configuration,userClassLoader)的构造方法，并传入了2个关键参数

 1.7、在父类StreamExecutionEnvironment(configuration,userClassLoader)的构造方法，又调用了StreamExecutionEnvironment(executorServiceLoader，configuration，userClassloader)，由2个关键参数变成了3个关键参数，即executorServiceLoader=new DefaultExecutorServiceLoader

 2、调用的是StreamExecutionEnvironment中第1696行的execute(String jobName)方法（注意返回值是JobExecutionResult类型），其内部调用的是：execute(getStreamGraph(jobName));

2.1、看一下，getStreamGraph(String jobName)方法的定义，其内部调用的是：getStreamGraph(jobName, true);

 

 

 2.2、查看getStreamGraph(jobName, clearTransformations)方法的定义，我们发现是通过getStreamGraphGenerator创建的StreamGraph

 

 

 2.3、查看getStreamGraphGenerator的定义：原来是new StreamGraphGenerator对象

3、在StreamExecutionEnvironment中，1696行的execute(String jobName)方法内部调用了位于1712行的execute(streamGraph)

 3.1、在StreamExecutionEnvironment中，1712行的execute(streamGraph)方法内部调用了executeAsync(streamGraph)方法，返回的是JobClient对象

 

 

 注意关键代码，

3.1.1、在1802~1803行：

final PipelineExecutorFactory executorFactory =
   executorServiceLoader.getExecutorFactory(configuration); // executorServiceLoader对象是DefaultExecutorServiceLoader类型，调用了getExecutorFactory(configuration)方法，

3.1.2、在1810~1812行：

CompletableFuture<JobClient> jobClientFuture = executorFactory
   .getExecutor(configuration)
   .execute(streamGraph, configuration);