flink standalone 客户端提交源码分析
2021/10/16 20:11:09
本文主要是介绍flink standalone 客户端提交源码分析,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
启动入口
CliFrontend.main -> cli.parseParameters -> ACTION_RUN run(params); -> executeProgram -> invokeInteractiveModeForExecution -> callMainMethod(){ mainMethod = entryClass.getMethod("main", String[].class); mainMethod.invoke(null, (Object) args); } --> SocketWindowWordCount.main(){ /************************************************* * TODO * 注释: 解析 host 和 port */ // 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; } /************************************************* * TODO * 注释: 获取 StreamExecutionEnvironment * 它呢,还是 跟 Spark 中的 SparkContext 还是有区别的! */ // get the execution environment final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); /************************************************* * TODO * 注释: 加载数据源得到数据抽象:DataStream * 其实最终,只是创建了一个 DataStreamSource 对象,然后把 SourceFunction(StreamOperator)和 StreamExecutionEnvironment * 设置到了 DataStreamSource 中, DataStreamSource 是 DataStream 的子类 * - * DataStream 的主要分类: * DataStreamSource 流数据源 * DataStreamSink 流数据目的地 * KeyedStream 按key分组的数据流 * DataStream 普通数据流 * - * 关于函数理解: * Function 传参 * Operator Graph 中抽象概念 * Transformation 一种针对流的逻辑操作 * 最终: Function ---> Operator ---> Transformation */ // 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 // TODO 注释: 讲算子生成 Transformation 加入到 Env 中的 transformations 集合中 .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)); } } }) // TODO 注释: 依然创建一个 DataStream(KeyedStream) .keyBy(value -> value.word) .timeWindow(Time.seconds(5)) // TODO 注释: .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); /************************************************* * TODO * 注释: 提交执行 */ 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<Transformation<?>> transformations 成员变量,该成员变量,主要用于保存 Job 的各种算子转化得到的 Transformation,把这些Transformation 按照逻辑拼接起来,就能得到 StreamGragh(Transformation ->StreamOperator -> StreamNode) 5、StreamExecutionEnvironment 提供了 execute() 方法主要用于提交 Job 执行。该方法接收的参数就是:StreamGraph --> env.socketTextStream -> addSource(){ /************************************************* * TODO * 注释: 获取输出数据类型 */ TypeInformation<OUT> resolvedTypeInfo = getTypeInfo(function, sourceName, SourceFunction.class, typeInfo); // TODO 注释: 判断是否是并行 boolean isParallel = function instanceof ParallelSourceFunction; clean(function); /************************************************* * TODO * 注释: 构建 SourceOperator * 它是 SourceFunction 的子类,也是 StreamOperator 的子类 */ final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function); /************************************************* * TODO * 注释: 返回 DataStreamSource * 关于这个东西的抽象有四种: * 1、DataStream * 2、KeyedDataStream * 3、DataStreamSource * 4、DataStreamSink */ return new DataStreamSource<>(this, resolvedTypeInfo, sourceOperator, isParallel, sourceName); } --> text.flatMap(讲算子生成 Transformation 加入到 Env 中的 transformations 集合中){ /************************************************* * TODO * 注释: 通过反射拿到 算子的类型 */ TypeInformation<R> outType = TypeExtractor.getFlatMapReturnTypes(); /************************************************* * TODO * 注释: 算子执行的真正操作逻辑是: 将算子构建成 Transformation 加入 Env 中的 transformation 中的 * transformations 集合中。将来执行 StreamGraph 生成的时候,会将 Transformation 变成 Operator * - * flatMap 到最后,还是构建一个 DataStream (SingleOutputStreamOperator)对象返回,然后将 Transformation 加入到 * transformations 集合中,等待将来提交的之后,构建成 StreamGraph */ return flatMap(flatMapper, outType); --> flatMap(){ /************************************************* * TODO * 注释: flink把每一个算子transform成一个对流的转换 * 并且注册到执行环境中,用于生成StreamGraph * - * 第一步:用户代码里定义的UDF会被当作其基类对待,然后交给 StreamFlatMap 这个 operator 做进一步包装。 * 事实上,每一个Transformation都对应了一个StreamOperator。 * - * flink流式计算的核心概念,就是将数据从输入流一个个传递给Operator进行链式处理,最后交给输出流的过程 * - * StreamFlatMap 是一个 Function 也是一个 StreamOperator * - * StreamFlatMap = StreamOperator * flatMapper = Function * -最终调用 transform 方法来把 StreamFlatMap 这种StreamOperator 转换成 Transformation * 最终加入到 StreamExectiionEnvironment 的 List<Transformation<?>> transformations */ 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(); /************************************************* * TODO * 注释: 构建: OneInputTransformation * 由于 flatMap 这个操作只接受一个输入,所以再被进一步包装为 OneInputTransformation */ OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(this.transformation, operatorName, operatorFactory, outTypeInfo,environment.getParallelism()); /************************************************* * TODO * 注释: 构建: SingleOutputStreamOperator */ @SuppressWarnings({"unchecked", "rawtypes"}) SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(environment, resultTransform); /************************************************* * TODO 重点 * 注释: 把 Operator 注册到执行环境中,用于生成 StreamGraph * 最后,将该 transformation 注册到执行环境中,当执行 generate 方法时,生成 StreamGraph 图结构。 */ getExecutionEnvironment().addOperator(resultTransform); /************************************************* * TODO * 注释: * SingleOutputStreamOperator 也是 DataStream 的子类,也就是返回了一个新的 DataStream * 然后调用新的 DataStream 的某一个算子,又生成新的 StreamTransformation, * 继续加入到 StreamExecutionEnvironment 的 transformations */ return returnStream; } } -> env.execute(提交执行)
StreamGraph
env.execute() -> StreamGraph sg = getStreamGraph(jobName); -> getStreamGraphGenerator().setJobName(jobName).generate(){ /************************************************* * TODO * 注释: 执行各种算子的 transformation: 由 算子 生成 Transformation 来构建 StreamGraph * 当时在执行各种算子的时候,就已经把算子转换成对应的 Transformation 放入 transformations 集合中了 * 自底向上(先遍历 input transformations) 对转换树的每个 transformation 进行转换 */ for(Transformation<?> transformation : transformations) { // TODO 注释: 从 Env 对象中,把 Transformation 拿出来,然后转换成 StreamNode // TODO 注释: Function --> Operator --> Transformation --> StreamNode transform(transformation); } --> transformOneInputTransform(){ // 1. 生成 streamNode /************************************************* * TODO * 注释: 添加一个 Operator(实际上 StreamGraph 端会添加一个 StreamNode) */ streamGraph .addOperator(transform.getId(), slotSharingGroup, transform.getCoLocationGroupKey(), transform.getOperatorFactory(), transform.getInputType(),transform.getOutputType(), transform.getName()); } -->addOperator -> addNode(){ /************************************************* * TODO * 注释: 生成一个 StreamNode */ StreamNode vertex = new StreamNode(vertexID, slotSharingGroup, coLocationGroup, operatorFactory, operatorName, new ArrayList<OutputSelector<?>>(),vertexClass); /************************************************* * TODO * 注释: 添加一个 StreamNode * * 这个 vertexID 就是 Transformation id transform.getId() * Transformation id 在创建 Transformation 时生成 * 即一个Transformation 对应一个 StreamNode */ streamNodes.put(vertexID, vertex); } // 2. 生成 StreamEdge /************************************************* * TODO * 注释: 定义该 SreamNode 的 入边 */ for(Integer inputId : inputIds) { /************************************************* * TODO * 注释: 设置当前 StreamNode 和 上游所有 StreamNode 之间的 StreamEdge * inputId 上游 * transform.getId() 下游 */ streamGraph.addEdge(inputId, transform.getId(), 0); } -> addEdge -> addEdgeInternal(){ /************************************************* * TODO * 注释: 构建 StreamNode 之间的 边(StreamEdge) 对象 */ 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(){ /************************************************* * TODO * 注释: 重点: 设置 Chaining, 将可以 Chain 到一起的 StreamNode Chain 在一起, * 这里会生成相应的 JobVertex 、JobEdge 、 IntermediateDataSet 对象 * 把能 chain 在一起的 Operator 都合并了,变成了 OperatorChain * - * 大致逻辑: * 这里的逻辑大致可以理解为,挨个遍历节点: * 1、如果该节点是一个 chain 的头节点,就生成一个 JobVertex, * 2、如果不是头节点,就要把自身配置并入头节点,然后把头节点和自己的出边相连; * 对于不能chain的节点,当作只有头节点处理即可 * - * 作用: * 能减少线程之间的切换,减少消息的序列化/反序列化,减少数据在缓冲区的交换,减少了延迟的同时提高整体的吞吐量。 */ setChaining(hashes, legacyHashes); // TODO 注释: 设置 PhysicalEdges // TODO 注释: 将每个 JobVertex 的入边集合也序列化到该 JobVertex 的 StreamConfig 中 // TODO 注释: 出边集合,已经在 上面的代码中,已经搞定了。 setPhysicalEdges(); // TODO 注释: 设置 SlotSharingAndCoLocation setSlotSharingAndCoLocation(); } --> setChaining(){ /************************************************* * TODO * 注释: * 1、一个 StreamNode 也可以认为是 做了 chain 动作 StreamNode -> JobVertex * 2、两个 StreamNode 做了 chain 动作 StreamNode + StreamNode -> JobVertex */ // TODO 注释: 处理每个 StreamNode for(Integer sourceNodeId : streamGraph.getSourceIDs()) { /************************************************* * TODO * 注释: 把能 chain 在一起的 StreamNode 都 chain 在一起 */ createChain(sourceNodeId, 0, new OperatorChainInfo(sourceNodeId, hashes, legacyHashes, streamGraph)); } } ---> createChain(){ /************************************************* * TODO -> * 注释: 判断是否可以 chain 在一起! * 当前这个地方做的事情,只是当前这个 StreamNode 和它的直接下游 StreamNode */ for(StreamEdge outEdge : currentNode.getOutEdges()) { /************************************************* * TODO 重点 1 isChainable * 注释: 判断一个 StreamGraph 中的一个 StreamEdge 链接的上下游 Operator(StreamNode) 是否可以 chain 在一起 * */ 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())); } /************************************************* * TODO -> 重点 2 createJobVertex * 注释: 把chain在一起的多个 Operator 创建成一个 JobVertex * 如果当前节点是 chain 的起始节点, 则直接创建 JobVertex 并返回 StreamConfig, 否则先创建一个空的 StreamConfig * createJobVertex 函数就是根据 StreamNode 创建对应的 JobVertex, 并返回了空的 StreamConfig * * --总结: * StreamGraph -> JobGrahph * 判断哪些StreamEge可以执行优化(chain),将 多个StreanNode 并成一个 JobVertex * * StreamNode_A -> (StreamNode_B -> StreamNode_C) * B,C chain在一起, startNodeId = B // Integer startNodeId = chainInfo.getStartNodeId(); * 当 currentNodeId = B 则 B 创建 JobVertex * 当 currentNodeId = C 则 C 不创建 JobVertex */ StreamConfig config = currentNodeId.equals(startNodeId) ? // TODO -> createJobVertex(startNodeId, chainInfo) : new StreamConfig(new Configuration()); // TODO 注释: chain 在一起的多条边 connect 在一起 for(StreamEdge edge : transitiveOutEdges) { /** * 重点 3 根据 StreamNode和 StreamEdge 生成 JobEge 和 IntermediateDataSet 用来将JobVertex和JobEdge相连 * TODO -> */ connect(startNodeId, edge); } } // 重点 1 isChainable isChainable(){ // TODO 注释: 获取上游 SourceVertex StreamNode upStreamVertex = streamGraph.getSourceVertex(edge); // TODO 注释: 获取下游 TargetVertex StreamNode downStreamVertex = streamGraph.getTargetVertex(edge); /************************************************* * TODO * 注释: 判断是否能 chain 在一起 */ // 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; } /************************************************* * 条件4、上游算子的链接策略是 always 或者 head * 条件5、下游算子的链接策略必须是 always */ 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){ /************************************************* * TODO * 注释: 先持久化: 把 JobGragh 持久化到磁盘文件形成 jobGraphFile * 1、持久化 JobGragh 的前缀:flink-jobgraph * 2、持久化 JobGragh 的后缀:.bin * 当我们把 JobGraph 持久化了之后,变成了一个文件: jobGraphFile * 然后其实,在提交 JobGraph 到 Flink 集群运行的时候,其实提交的就是: 这个文件! * 将来,最终是有 FLink 集群的 WebMonitor(JobSubmitHandler) 去接收请求来执行处理 * JobSubmitHandler 在执行处理的第一件事情: 把传送过来的这个文件反序列化得到 JobGraph 这个对象 */ CompletableFuture<java.nio.file.Path> 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); /************************************************* * TODO * 注释: 等待持久化完成之后,然后加入带上传文件系列 * 补充: thenApply 接收一个函数作为参数,使用该函数处理上一个 CompletableFuture 调用的结果,并返回一个具有处理结果的 Future 对象。 */ CompletableFuture<Tuple2<JobSubmitRequestBody, Collection<FileUpload>>> requestFuture = jobGraphFileFuture.thenApply(jobGraphFile -> { List<String> jarFileNames = new ArrayList<>(8); List<JobSubmitRequestBody.DistributedCacheFile> artifactFileNames = new ArrayList<>(8); Collection<FileUpload> 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(){ /************************************************* * TODO * 注释: 通过 Netty 客户端发送请求给 Netty 服务端 */ final ChannelFuture connectFuture = bootstrap.connect(targetAddress, targetPort); /************************************************* * TODO -> * 注释: 发送请求 到 WebMonitorEndpoint 的 Netty 服务端 * 最终由: JobSubmitHandler 来执行处理 */ httpRequest.writeTo(channel); }
这篇关于flink standalone 客户端提交源码分析的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-11-26Mybatis官方生成器资料详解与应用教程
- 2024-11-26Mybatis一级缓存资料详解与实战教程
- 2024-11-26Mybatis一级缓存资料详解:新手快速入门
- 2024-11-26SpringBoot3+JDK17搭建后端资料详尽教程
- 2024-11-26Springboot单体架构搭建资料:新手入门教程
- 2024-11-26Springboot单体架构搭建资料详解与实战教程
- 2024-11-26Springboot框架资料:新手入门教程
- 2024-11-26Springboot企业级开发资料入门教程
- 2024-11-26SpringBoot企业级开发资料详解与实战教程
- 2024-11-26Springboot微服务资料:新手入门全攻略