启动入口
CliFrontend.main -> cli.parseParameters -> ACTION_RUN run(params); -> executeProgram -> invokeInteractiveModeForExecution
-> callMainMethod(){
mainMethod = entryClass.getMethod("main", String[].class);
mainMethod.invoke(null, (Object) args);
}
--> SocketWindowWordCount.main(){
// the host and the port to connect to
final String hostname;
final int port;
try {
final ParameterTool params = ParameterTool.fromArgs(args);
hostname = params.has("hostname") ? params.get("hostname") : "localhost";
port = params.getInt("port");
} catch(Exception e) {
return;
}
// get the execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// get input data by connecting to the socket
DataStream text = env.socketTextStream(hostname, port, "n");
// parse the data, group it, window it, and aggregate the counts
DataStream windowCounts = text
// TODO 注释： 讲算子生成 Transformation 加入到 Env 中的 transformations 集合中
.flatMap(new FlatMapFunction() {
@Override
public void flatMap(String value, Collector out) {
for(String word : value.split("\s")) {
out.collect(new WordWithCount(word, 1L));
}
}
})
// TODO 注释： 依然创建一个 DataStream(KeyedStream)
.keyBy(value -> value.word)
.timeWindow(Time.seconds(5))
// TODO 注释：
.reduce(new ReduceFunction() {
@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");
}
--> StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamExecutionEnvironment 是 Flink 应用程序的执行入口，提供了一些重要的操作机制：
1、提供了 readTextFile(), socketTextStream(), createInput(), addSource() 等方法去对接数据源
2、提供了 setParallelism() 设置程序的并行度
3、StreamExecutionEnvironment 管理了 ExecutionConfig 对象，该对象负责Job执行的一些行为配置管理。
还管理了 Configuration 管理一些其他的配置
4、StreamExecutionEnvironment 管理了一个 List> transformations
成员变量，该成员变量，主要用于保存 Job 的各种算子转化得到的 Transformation，把这些Transformation 按照逻辑拼接起来，就能得到 StreamGragh（Transformation ->StreamOperator -> StreamNode）
5、StreamExecutionEnvironment 提供了 execute() 方法主要用于提交 Job 执行。该方法接收的参数就是：StreamGraph
--> env.socketTextStream -> addSource(){
TypeInformation resolvedTypeInfo = getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);
// TODO 注释： 判断是否是并行
boolean isParallel = function instanceof ParallelSourceFunction;
clean(function);
final StreamSource sourceOperator = new StreamSource<>(function);
return new DataStreamSource<>(this, resolvedTypeInfo, sourceOperator, isParallel, sourceName);
}
--> text.flatMap(讲算子生成 Transformation 加入到 Env 中的 transformations 集合中){
TypeInformation outType = TypeExtractor.getFlatMapReturnTypes();
return flatMap(flatMapper, outType);
--> flatMap(){
return transform("Flat Map", outputType, new StreamFlatMap<>(clean(flatMapper)));
}
---> doTransform(){
// read the output type of the input Transform to coax out errors about MissingTypeInfo
transformation.getOutputType();
OneInputTransformation resultTransform = new OneInputTransformation<>(this.transformation, operatorName, operatorFactory, outTypeInfo,environment.getParallelism());
@SuppressWarnings({"unchecked", "rawtypes"}) SingleOutputStreamOperator returnStream = new SingleOutputStreamOperator(environment, resultTransform);
getExecutionEnvironment().addOperator(resultTransform);
return returnStream;
}
}
-> env.execute(提交执行)
StreamGraph
env.execute() -> StreamGraph sg = getStreamGraph(jobName); -> getStreamGraphGenerator().setJobName(jobName).generate(){
for(Transformation transformation : transformations) {
// TODO 注释： 从 Env 对象中，把 Transformation 拿出来，然后转换成 StreamNode
// TODO 注释： Function --> Operator --> Transformation --> StreamNode
transform(transformation);
}
--> transformOneInputTransform(){
// 1. 生成 streamNode
streamGraph
.addOperator(transform.getId(), slotSharingGroup, transform.getCoLocationGroupKey(), transform.getOperatorFactory(), transform.getInputType(),transform.getOutputType(), transform.getName());
}
-->addOperator -> addNode(){
StreamNode vertex = new StreamNode(vertexID, slotSharingGroup, coLocationGroup, operatorFactory, operatorName, new ArrayList>(),vertexClass);
streamNodes.put(vertexID, vertex);
}
// 2. 生成 StreamEdge
for(Integer inputId : inputIds) {
streamGraph.addEdge(inputId, transform.getId(), 0);
}
-> addEdge -> addEdgeInternal(){
StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner, outputTag, shuffleMode);
// TODO 注释： 给 上游 StreamNode 设置 出边
getStreamNode(edge.getSourceId()).addOutEdge(edge);
// TODO 注释： 给 下游 StreamNode 设置 入边
getStreamNode(edge.getTargetId()).addInEdge(edge);
}
}
总结：
1、生成上游顶点和下游顶点 StreamNode upstreamNode | StreamNode downstreamNode
2、根据上下游顶点生成 StreamEdge StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode...)
3、将成的StreamEdge 加入上游StreamNode 的 出边 getStreamNode(edge.getSourceId()).addOutEdge(edge);
为啥不直接用 upstreamNode.addOutEdge(edge);
4、将成的StreamEdge 加入下游StreamNode 的 入边 getStreamNode(edge.getTargetId()).addInEdge(edge);
JobGraph
execute(sg); -> executeAsync -> AbstractSessionClusterExecutor.execute(){
// 1. 将streamgraph优化得到jobgraph
final JobGraph jobGraph = PipelineExecutorUtils.getJobGraph(pipeline, configuration);
// 2. 调用RestClient中的netty 客户端进行提交 到 服务端执行
// 通过 channel 把请求数据，发给 WebMonitorEndpoint 中的 JobSubmitHandler 来执行处理
clusterClient.submitJob(jobGraph)
}
// 1. 将streamgraph优化得到jobgraph
-> PipelineExecutorUtils.getJobGraph(pipeline, configuration) -> FlinkPipelineTranslationUtil.getJobGraph
-> pipelineTranslator.translateToJobGraph -> streamGraph.getJobGraph -> StreamingJobGraphGenerator.createJobGraph
-> new StreamingJobGraphGenerator(streamGraph, jobID).createJobGraph(){
setChaining(hashes, legacyHashes);
// TODO 注释： 设置 PhysicalEdges
// TODO 注释： 将每个 JobVertex 的入边集合也序列化到该 JobVertex 的 StreamConfig 中
// TODO 注释： 出边集合，已经在 上面的代码中，已经搞定了。
setPhysicalEdges();
// TODO 注释： 设置 SlotSharingAndCoLocation
setSlotSharingAndCoLocation();
}
--> setChaining(){
// TODO 注释： 处理每个 StreamNode
for(Integer sourceNodeId : streamGraph.getSourceIDs()) {
createChain(sourceNodeId, 0, new OperatorChainInfo(sourceNodeId, hashes, legacyHashes, streamGraph));
}
}
---> createChain(){
for(StreamEdge outEdge : currentNode.getOutEdges()) {
if(isChainable(outEdge, streamGraph)) {
// TODO 注释： 加入可 chain 集合
chainableOutputs.add(outEdge);
} else {
// TODO 注释： 加入不可 chain 集合
nonChainableOutputs.add(outEdge);
}
}
// TODO 注释： 把可以 chain 在一起的 StreamEdge 两边的 Operator chain 在一个形成一个 OperatorChain
for(StreamEdge chainable : chainableOutputs) {
// TODO 注释： 递归 chain
// TODO 注释： 如果可以 chain 在一起的话，这里的 chainIndex 会加 1
transitiveOutEdges.addAll(createChain(chainable.getTargetId(), chainIndex + 1, chainInfo));
}
// 不能chain在一起的
for(StreamEdge nonChainable : nonChainableOutputs) {
transitiveOutEdges.add(nonChainable);
// TODO 注释： 不能 chain 一起的话，这里的 chainIndex 是从 0 开始算的，后面也肯定会走到 createJobVertex 的逻辑
createChain(nonChainable.getTargetId(), 0, chainInfo.newChain(nonChainable.getTargetId()));
}
StreamConfig config = currentNodeId.equals(startNodeId) ?
// TODO ->
createJobVertex(startNodeId, chainInfo) : new StreamConfig(new Configuration());
// TODO 注释： chain 在一起的多条边 connect 在一起
for(StreamEdge edge : transitiveOutEdges) {
connect(startNodeId, edge);
}
}
// 重点 1 isChainable
isChainable(){
// TODO 注释： 获取上游 SourceVertex
StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
// TODO 注释： 获取下游 TargetVertex
StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);
// TODO 条件1. 下游节点的入度为1 （也就是说下游节点没有来自其他节点的输入） A -> B B A 一一对应 如果shuffle类，那么B的入度就 >= 2
return downStreamVertex.getInEdges().size() == 1
// TODO 注释： 条件2. 上下游算子实例处于同一个SlotSharingGroup中
&& upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
// TODO -> 注释： 这里面有 3 个条件 条件 345
&& areOperatorsChainable(upStreamVertex, downStreamVertex, streamGraph){
// TODO 注释： 获取 上游 Operator
StreamOperatorFactory upStreamOperator = upStreamVertex.getOperatorFactory();
// TODO 注释： 获取 下游 Operator
StreamOperatorFactory downStreamOperator = downStreamVertex.getOperatorFactory();
// TODO 注释：条件3、前后算子不为空 如果上下游有一个为空，则不能进行 chain
if(downStreamOperator == null || upStreamOperator == null) {
return false;
}
if(upStreamOperator.getChainingStrategy() == ChainingStrategy.NEVER ||
downStreamOperator.getChainingStrategy() != ChainingStrategy.ALWAYS) {
return false;
}
}
// TODO 注释：条件6 两个算子间的物理分区逻辑是ForwardPartitioner
// (无shuffle，当前节点的计算数据，只会发给自己 one to one 如上游50个task 计算完直接发送给下游50个task)
&& (edge.getPartitioner() instanceof ForwardPartitioner)
// TODO 注释：条件7 两个算子间的shuffle方式不等于批处理模式
&& edge.getShuffleMode() != ShuffleMode.BATCH
// TODO 注释：条件8 上下游算子实例的并行度相同
&& upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
// TODO 注释：条件9 启动了 chain
&& streamGraph.isChainingEnabled();
}
// 重点 2 createJobVertex
createJobVertex(){
// TODO 注释： 获取 startStreamNode
StreamNode streamNode = streamGraph.getStreamNode(streamNodeId);
// TODO 注释： 生成一个 JobVertexID
JobVertexID jobVertexId = new JobVertexID(hash);
// JobVertex 初始化
if(chainedInputOutputFormats.containsKey(streamNodeId)) {
jobVertex = new InputOutputFormatVertex(chainedNames.get(streamNodeId), jobVertexId, operatorIDPairs);
chainedInputOutputFormats.get(streamNodeId).write(new TaskConfig(jobVertex.getConfiguration()));
} else {
// TODO 注释： 创建一个 JobVertex
jobVertex = new JobVertex(chainedNames.get(streamNodeId), jobVertexId, operatorIDPairs);
}
// 将生成好的 JobVertex 加入到： JobGraph
jobGraph.addVertex(jobVertex);
}
// 重点 3 根据 StreamNode和 StreamEdge 生成 JobEge 和 IntermediateDataSet 用来将JobVertex和JobEdge相连
connect(startNodeId, edge){
//生成JobEdge
JobEdge jobEdge;
if(isPointwisePartitioner(partitioner)) {
jobEdge = downStreamVertex.connectNewDataSetAsInput(headVertex, DistributionPattern.POINTWISE, resultPartitionType);
} else {
// TODO -> 创建 IntermediateDataSet
jobEdge = downStreamVertex.connectNewDataSetAsInput(headVertex, DistributionPattern.ALL_TO_ALL, resultPartitionType);
}
}
----> connectNewDataSetAsInput(){
// TODO -> input是JobVertex 即 JobVertex 创建 IntermediateDataSet
IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType);
// TODO 创建 JobEdge
JobEdge edge = new JobEdge(dataSet, this, distPattern);
this.inputs.add(edge);
// TODO IntermediateDataSet -> JobEdge
dataSet.addConsumer(edge);
return edge;
// 至此形成流图 JobVertex -> IntermediateDataSet -> JobEdge
}
==================================================================================================================================
// 2. 调用RestClient中的netty 客户端进行提交 到 服务端执行
-> clusterClient.submitJob(jobGraph){
CompletableFuture jobGraphFileFuture = CompletableFuture.supplyAsync(() -> {
try {
final java.nio.file.Path jobGraphFile = Files.createTempFile("flink-jobgraph", ".bin");
try(ObjectOutputStream objectOut = new ObjectOutputStream(Files.newOutputStream(jobGraphFile))) {
objectOut.writeObject(jobGraph);
CompletableFuture>> requestFuture = jobGraphFileFuture.thenApply(jobGraphFile -> {
List jarFileNames = new ArrayList<>(8);
List artifactFileNames = new ArrayList<>(8);
Collection filesToUpload = new ArrayList<>(8);
// TODO 注释： 加入待上传的文件系列
filesToUpload.add(new FileUpload(jobGraphFile, RestConstants.CONTENT_TYPE_BINARY));
for(Path jar : jobGraph.getUserJars()) {
jarFileNames.add(jar.getName());
// 上传
filesToUpload.add(new FileUpload(Paths.get(jar.toUri()), RestConstants.CONTENT_TYPE_JAR));
}
// -> TODO -> 注释：sendRetriableRequest() 提交 真正提交
requestAndFileUploads -> sendRetriableRequest(JobSubmitHeaders.getInstance(), EmptyMessageParameters.getInstance(), requestAndFileUploads.f0,
requestAndFileUploads.f1, isConnectionProblemOrServiceUnavailable()));
// TODO 注释： 等 sendRetriableRequest 提交完成之后，删除生成的 jobGraghFile
Files.delete(jobGraphFile);
}
--> sendRetriableRequest -> restClient.sendRequest(){
final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort);
httpRequest.writeTo(channel);
}
ExecutionGraph
接上文
httpRequest.writeTo(channel);
发送请求 到 WebMonitorEndpoint 的 Netty 服务端 最终 JobSubmitHandler 来执行处理
具体参考 2.1 启动 webMonitorEndpoint 源码分析
JobSubmitHandler.handleRequest() -> DispatcherGateway.submitJob -> internalSubmitJob -> persistAndRunJob
-> runJob(){
final CompletableFuture jobManagerRunnerFuture = createJobManagerRunner(jobGraph);
FunctionUtils.uncheckedFunction(this::startJobManagerRunner)
}
// 重点1：createJobManagerRunner 创建 JobMaster , 将JobGraph 转换成 ExecutionGraph
createJobManagerRunner -> createJobManagerRunner -> new JobManagerRunnerImpl(负责启动 JobMaster)
-> jobMasterFactory.createJobMasterService -> new JobMaster
-> this.schedulerNG = createScheduler(jobManagerJobMetricGroup); -> createInstance -> new DefaultScheduler
-> super -> Schedulerbase(){
this.executionGraph = createAndRestoreExecutionGraph(jobManagerJobMetricGroup, checkNotNull(shuffleMaster), checkNotNull(partitionTracker));
-> ExecutionGraph newExecutionGraph = createExecutionGraph(currentJobManagerJobMetricGroup, shuffleMaster, partitionTracker);
}
--> createExecutionGraph() -> ExecutionGraphBuilder.buildGraph(){
executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));
executionGraph.attachJobGraph(sortedTopology);
} -> attachJobGraph(){
for(JobVertex jobVertex : topologiallySorted) {
if(jobVertex.isInputVertex() && !jobVertex.isStoppable()) {
this.isStoppable = false;
}
// create the execution job vertex and attach it to the graph
ExecutionJobVertex ejv = new ExecutionJobVertex(this, jobVertex, 1, maxPriorAttemptsHistoryLength, rpcTimeout, globalModVersion, createTimestamp);
ejv.connectToPredecessors(this.intermediateResults);
ExecutionJobVertex previousTask = this.tasks.putIfAbsent(jobVertex.getID(), ejv);
if(previousTask != null) {
throw new JobException(
String.format("Encountered two job vertices with ID %s : previous=[%s] / new=[%s]", jobVertex.getID(), ejv, previousTask));
}
for(IntermediateResult res : ejv.getProducedDataSets()) {
IntermediateResult previousDataSet = this.intermediateResults.putIfAbsent(res.getId(), res);
if(previousDataSet != null) {
throw new JobException(
String.format("Encountered two intermediate data set with ID %s : previous=[%s] / new=[%s]", res.getId(), res, previousDataSet));
}
}
this.verticesInCreationOrder.add(ejv);
// TODO 注释： 总并行度
this.numVerticesTotal += ejv.getParallelism();
// TODO 注释： ExecutionJobVertex 加入 newExecJobVertices List 中
newExecJobVertices.add(ejv);
}
}
--> ejv.connectToPredecessors(this.intermediateResults){
for(int num = 0; num < inputs.size(); num++) {
// TODO 注释： 遍历到一个 JobEdge
JobEdge edge = inputs.get(num);
// TODO 注释： 获取到 JobEdge 链接的 IntermediateResult
IntermediateResult ires = intermediateDataSets.get(edge.getSourceId());
// TODO 注释： 将当前 IntermediateResult 作为 ExecutionJobVertex 的输入
// TODO 注释： 加入 inputs 集合，作为 ExecutionJobVertex 的输入
this.inputs.add(ires);
for(int i = 0; i < parallelism; i++) {
ExecutionVertex ev = taskVertices[i];
ev.connectSource(num, ires, edge, consumerIndex);
}
}
}
----> connectSource(){
edges = connectAllToAll(sourcePartitions, inputNumber){
for(int i = 0; i < sourcePartitions.length; i++) {
IntermediateResultPartition irp = sourcePartitions[i];
edges[i] = new ExecutionEdge(irp, this, inputNumber);
}
return edges;
}
}
// 重点2：startJobManagerRunner 启动 JobMaster
startJobManagerRunner -> jobManagerRunner.start(); -> leaderElectionService.start(this); -> grantLeadership
-> verifyJobSchedulingStatusAndStartJobManager -> startJobMaster -> jobMasterService.start
-> startJobExecution(){
startJobMasterServices();
resetAndStartScheduler();
}
JobMaster 向 ResourceManager 和 TaskManager 注册和维持心跳
startJobMasterServices() -> startHeartbeatServices()
private void startHeartbeatServices() {
taskManagerHeartbeatManager = heartbeatServices
.createHeartbeatManagerSender(resourceId, new TaskManagerHeartbeatListener(), getMainThreadExecutor(), log);
resourceManagerHeartbeatManager = heartbeatServices
.createHeartbeatManager(resourceId, new ResourceManagerHeartbeatListener(), getMainThreadExecutor(), log);
}
reconnectToResourceManager -> tryConnectToResourceManager -> connectToResourceManager
-> resourceManagerConnection.start(){
// 创建注册
createNewRegistration() -> generateRegistration
// 开始注册
newRegistration.startRegistration() -> register -> invokeRegistration -> registerJobManager
-> registerJobMasterInternal(){
JobManagerRegistration jobManagerRegistration = new JobManagerRegistration(jobId, jobManagerResourceId, jobMasterGateway);
jobManagerRegistrations.put(jobId, jobManagerRegistration);
jmResourceIdRegistrations.put(jobManagerResourceId, jobManagerRegistration);
// 维持心跳
jobManagerHeartbeatManager.monitorTarget(){
public void requestHeartbeat(ResourceID resourceID, Void payload) {
jobMasterGateway.heartbeatFromResourceManager(resourceID);
}
}
}
resourceManagerLeaderRetriever.start(new ResourceManagerLeaderListener());
JobMaster 开始申请 Slot，并且部署 Task
resetAndStartScheduler -> startScheduling
-> startAllOperatorCoordinators
-> startSchedulingInternal(){
schedulingStrategy.startScheduling(); -> allocateSlotsAndDeploy(){
final List slotExecutionVertexAssignments = allocateSlots(executionVertexDeploymentOptions);
waitForAllSlotsAndDeploy(deploymentHandles);
}
}
Slot 管理（申请和释放）源码解析
allocateSlots
Task 部署和提交
waitForAllSlotsAndDeploy
