当前位置： 首页 > article >正文

Flink源码解析之：如何根据StreamGraph生成JobGraph

article 2025/1/5 6:17:39

Flink源码解析之：如何根据StreamGraph生成JobGraph

在上一章节中，我们讲解了Flink如何将用户自定义逻辑算子转换成StreamGraph。在生成StreamGraph的过程中，Flink内部没有做任何优化，只是将用户自定义算子和处理流程转换成了StreamGraph的拓扑图来表示。本章我们将会介绍StreamGraph到JobGraph的生成流程，在这一过程中，**Flink内部是做了优化操作的，主要是做了算子的Chain操作，Chain在一起的算子会形成算子链在同一个线程上下文中执行，减少算子间的上下文切换开销以及shuffle开销。**接下来，我们就带大家通过源码来深入理解这一转换流程，并理解Flink优化的算子Chain操作原理。

源码入口与转换图

StreamGraph -> JobGraph的转换核心源码，在包org.apache.flink.streaming.api.graph 下的StreamingJobGraphGenerator，该类的注解也表明了该类的作用：

The StreamingJobGraphGenerator converts a {@link StreamGraph} into a {@link JobGraph}.

StreamGraph的整体转换流程可以参考下图，有一个大致的概念，在阅读完本文后再回来看这张图可能会有更深刻的理解。

在这里插入图片描述

具体实现原理

首先进入到StreamingJobGraphGenerator类的转换入口方法createJobGraph中，源码如下所示：

private JobGraph createJobGraph() {
	// 校验应用参数
    preValidate();
    jobGraph.setJobType(streamGraph.getJobType());

    jobGraph.enableApproximateLocalRecovery(
            streamGraph.getCheckpointConfig().isApproximateLocalRecoveryEnabled());

    // Generate deterministic hashes for the nodes in order to identify them across
    // submission iff they didn't change.
	// note: 为每个StreamNode生成一个确定的hash id，如果提交的拓扑结构没有改变，则每次生成的hash id都会是一样的
	// note: 此哈希码基于节点的当前状态以及其所有输入的状态。
    Map<Integer, byte[]> hashes =
            defaultStreamGraphHasher.traverseStreamGraphAndGenerateHashes(streamGraph);

    // Generate legacy version hashes for backwards compatibility
	// note: 这个设置主要是为了防止 hash 机制变化时出现不兼容的情况
    List<Map<Integer, byte[]>> legacyHashes = new ArrayList<>(legacyStreamGraphHashers.size());
    for (StreamGraphHasher hasher : legacyStreamGraphHashers) {
        legacyHashes.add(hasher.traverseStreamGraphAndGenerateHashes(streamGraph));
    }

	// note: 转换中最重要的方法，合并算子形成算子链、生成jobvertex、连接算子顶点等操作
    setChaining(hashes, legacyHashes);

	// note: note: 将每个 JobVertex 的入边集合也序列化到该 JobVertex 的 StreamConfig 中 (出边集合已经在 setChaining 的时候写入了)
    setPhysicalEdges();

	// note: 为每个 JobVertex 指定所属的 SlotSharingGroup 以及设置 CoLocationGroup
    setSlotSharingAndCoLocation();

    setManagedMemoryFraction(
            Collections.unmodifiableMap(jobVertices),
            Collections.unmodifiableMap(vertexConfigs),
            Collections.unmodifiableMap(chainedConfigs),
            id -> streamGraph.getStreamNode(id).getManagedMemoryOperatorScopeUseCaseWeights(),
            id -> streamGraph.getStreamNode(id).getManagedMemorySlotScopeUseCases());

	// note: checkpoint相关的配置
    configureCheckpointing();

    // note: 用户的第三方依赖包就是在这里（cacheFile）传给 JobGraph
jobGraph.setSavepointRestoreSettings(streamGraph.getSavepointRestoreSettings());

    final Map<String, DistributedCache.DistributedCacheEntry> distributedCacheEntries =
            JobGraphUtils.prepareUserArtifactEntries(
                    streamGraph.getUserArtifacts().stream()
                            .collect(Collectors.toMap(e -> e.f0, e -> e.f1)),
                    jobGraph.getJobID());

    for (Map.Entry<String, DistributedCache.DistributedCacheEntry> entry :
            distributedCacheEntries.entrySet()) {
        jobGraph.addUserArtifact(entry.getKey(), entry.getValue());
    }

    // set the ExecutionConfig last when it has been finalized
    try {
		//note: 将 StreamGraph 的 ExecutionConfig 序列化到 JobGraph 的配置中
        jobGraph.setExecutionConfig(streamGraph.getExecutionConfig());
    } catch (IOException e) {
        throw new IllegalConfigurationException(
                "Could not serialize the ExecutionConfig."
                        + "This indicates that non-serializable types (like custom serializers) were registered");
    }

    return jobGraph;
}

该方法的主要实现步骤如下：

  • 首先，在做任何操作之前，它运行preValidate()来验证流图的状态，确保其有效性。
  • 使用traverseStreamGraphAndGenerateHashes 方法生成流图中每个节点的哈希值。此哈希码基于节点的当前状态以及其所有输入的状态。这是为了实现Flink的状态后端，特别是在恢复作业/从保存点启动作业时需要使用这些哈希码来准确找到并恢复节点的状态。
  • setChaining(hashes, legacyHashes) 调用用于为StreamGraph设置操作链。链式操作可以在单个任务中执行多个操作符，以减少数据在任务之间的传输，以此提高性能。
    这里会生成相应的 JobVertex 、JobEdge 、 IntermediateDataSet 对象，JobGraph 的 Graph 在这一步就已经完全构建出来了；
  • setPhysicalEdges() `方法会将每个 JobVertex 的入边集合也序列化到该 JobVertex 的 StreamConfig 中 (出边集合已经在 setChaining 的时候写入了)；
  • setSlotSharingAndCoLocation()设置插槽共享和共位，以实现任务之间的资源共享。
  • setManagedMemoryFraction() 设置管理内存的分数，参数包括jobVertices, vertexConfigs, chainedConfigs，以及一些函数，用于根据id获取流节点的权重和使用案例。
  • configureCheckpointing() 配置检查点。检查点是Flink提供的容错机制，可以恢复到特定状态以保证结果的正确性。
  • 最后，将StreamGraph中的ExecutionConfig（执行配置）序列化并设置给JobGraph。如果ExecutionConfig不能被序列化，例如其中含有自定义序列化程序，那么将抛出IllegalConfigurationException异常。

上面的方法中，最重要的就是setChaining方法，该方法中设置了算子的操作链，并生成JobVertex、JobEdge、IntermediateDataSet对象，并将它们连接成JobGraph的拓扑图。

为此，接下来，我们会核心着重讲解该方法。

我们先进入setChaining方法中，探究其源码执行原理。

/**
 * Sets up task chains from the source {@link StreamNode} instances.
 *
 * <p>This will recursively create all {@link JobVertex} instances.
 */
private void setChaining(Map<Integer, byte[]> hashes, List<Map<Integer, byte[]>> legacyHashes) {
    // we separate out the sources that run as inputs to another operator (chained inputs)
    // from the sources that needs to run as the main (head) operator.
    final Map<Integer, OperatorChainInfo> chainEntryPoints =
            buildChainedInputsAndGetHeadInputs(hashes, legacyHashes);
    final Collection<OperatorChainInfo> initialEntryPoints =
            chainEntryPoints.entrySet().stream()
                    .sorted(Comparator.comparing(Map.Entry::getKey))
                    .map(Map.Entry::getValue)
                    .collect(Collectors.toList());

    // iterate over a copy of the values, because this map gets concurrently modified
    for (OperatorChainInfo info : initialEntryPoints) {
        createChain(
                info.getStartNodeId(),
                1, // operators start at position 1 because 0 is for chained source inputs
                info,
                chainEntryPoints);
    }
}

上述代码首先执行了buildChainedInputsAndGetHeadInputs方法。该方法有什么作用呢？为了节省篇幅，暂且不列出该方法的源码，只分析其实现原理和作用。以下是该方法的主要步骤：

  1. 该方法内部会遍历所有的源节点，并判断源节点与下游直接连接的节点是否可以合并成一个算子链。
  2. 如果可以合并，则会为当前源节点创建一个StreamConfig对象，并配置源节点在当前算子链中的索引（设置为0，因为是开头）、算子ID、算子名称等元数据信息，并添加到一个chainedSources的Map对象中，该对象包含了每个源节点的上述元数据信息。
  3. 如果可以合并，则会为源节点的下游节点创建一个算子链对象OperatorChainInfo，该对象会包含上面的chainedSources属性，以标识该算子的算子链源节点。并将该下游节点和OperatorChainInfo对象放入chainEntryPoints的Map结构中。
  4. 如果无法合并，则只会将源节点本身创建算子链OperatorChainInfo对象，放入chainEntryPoints的Map结构中。表明从源头的初始算子链，只有其本身。

经过上面的处理后，就可以得到从源节点开始的初始算子链，只包含源节点和其下游节点。chainEntryPoints 映射将包含所有源节点的 OperatorChainInfo，以及可被链接的源节点的紧邻下游节点信息。其中那些可以被链到其他操作符的源节点信息也同时存储在 chainedSources映射中。这个映射将被用于后续的处理。

以下图的StreamGraph拓扑图为例，直观来看，chainedSources对象存储的是可被chain的op1、op2源节点的信息。chainEntryPoint存储的是op1、op2、op3、op4，包含源节点及其可链接的邻接下游节点信息。

在这里插入图片描述

有了上述的源节点可链接信息后，便从源节点开始遍历chainEntryPoints的values集合，对每个源节点执行createChain方法。该方法中会递归地对源节点的所有下游节点进行遍历，以判断哪些是可以链接的，哪些是不可链接的，并最终生成JobVertex和JobEdge进行连接。接下来，我们着重来分析该方法流程。

算子是如何Chain在一起的

这一小节，我们来介绍生成JobGraph的一个核心步骤，即算子是如何Chain到一起的。在具体讲解前，先看一下算子Chain的示例图：

在这里插入图片描述

上图可以看到，在StreamGraph中，从KeyedAggregation算子到DataSink算子是forward的分区方式，当Flink判断两者是可以链接到一起时，便会在转换成JobGraph时，将两个算子合并在一个算子链中，生成一个JobVertex。

StreamGraph转换为JobGraph的处理过程主要是在createChain方法中完成的，先来看下这个方法的实现：

private List<StreamEdge> createChain(
        final Integer currentNodeId,
        final int chainIndex,
        final OperatorChainInfo chainInfo,
        final Map<Integer, OperatorChainInfo> chainEntryPoints) {

	// 算子链的起始节点
    Integer startNodeId = chainInfo.getStartNodeId();
    if (!builtVertices.contains(startNodeId)) {

        List<StreamEdge> transitiveOutEdges = new ArrayList<StreamEdge>();

		// 记录能链接在一起的下游边集合和不能链接在一起的下游边集合
        List<StreamEdge> chainableOutputs = new ArrayList<StreamEdge>();
        List<StreamEdge> nonChainableOutputs = new ArrayList<StreamEdge>();

		// 算子链中当前要处理的StreamNode
        StreamNode currentNode = streamGraph.getStreamNode(currentNodeId);

		// 遍历当前节点的下游节点，判断是否可以chain在一起，并把出边写入相应集合中
        for (StreamEdge outEdge : currentNode.getOutEdges()) {
            if (isChainable(outEdge, streamGraph)) {
                chainableOutputs.add(outEdge);
            } else {
                nonChainableOutputs.add(outEdge);
            }
        }

		// 对于能够链接的下游节点，递归继续遍历其出边，重复上面的逻辑。
		// chainIndex表示当前StreamNode在算子链中的索引位置，递归下游节点时需要+1
        for (StreamEdge chainable : chainableOutputs) {
            transitiveOutEdges.addAll(
                    createChain(
                            chainable.getTargetId(),
                            chainIndex + 1,
                            chainInfo,
                            chainEntryPoints));
        }

		// 遍历不可链接的出边集合，并加入到transitiveOutEdges集合中
		// 这个transitiveOutEdges会在递归调用中返回，调用栈中上层调用会使用下面的connect方法将当前StreamNode与下层调用返回的transitiveOutEdges中的出边进行连接（如果不为空的话）
		// 表明当前StreamNode和返回的下游算子链顶点不能合并在同一个算子链中
        for (StreamEdge nonChainable : nonChainableOutputs) {
            transitiveOutEdges.add(nonChainable);
			// 下游StreamNode会成为新的算子链顶点，只不过这里chainIndex会设置为1
			// 并继续往下递归，构造以该下游StreamNode为顶点开始的算子链
            createChain(
                    nonChainable.getTargetId(),
                    1, // operators start at position 1 because 0 is for chained source inputs
                    chainEntryPoints.computeIfAbsent(
                            nonChainable.getTargetId(),
                            (k) -> chainInfo.newChain(nonChainable.getTargetId())),
                    chainEntryPoints);
        }

		// 记录 chainedName
        chainedNames.put(
                currentNodeId,
                createChainedName(
                        currentNodeId,
                        chainableOutputs,
                 Optional.ofNullable(chainEntryPoints.get(currentNodeId))));
		
		// 计算Chain之后，算子链的minResources
        chainedMinResources.put(
                currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
		// 计算chain之后，算子链的资源上限
        chainedPreferredResources.put(
                currentNodeId,
                createChainedPreferredResources(currentNodeId, chainableOutputs));

        OperatorID currentOperatorId =
                chainInfo.addNodeToChain(
                        currentNodeId,
                        streamGraph.getStreamNode(currentNodeId).getOperatorName());

        if (currentNode.getInputFormat() != null) {
            getOrCreateFormatContainer(startNodeId)
                    .addInputFormat(currentOperatorId, currentNode.getInputFormat());
        }

        if (currentNode.getOutputFormat() != null) {
            getOrCreateFormatContainer(startNodeId)
                    .addOutputFormat(currentOperatorId, currentNode.getOutputFormat());
        }

		// 如果当前节点和startNodeId一致，说明递归过程回到了算子链的起始节点，则直接创建JobVertex，否则先创建一个空的StreamConfig
		// createJobVertex方法就是根据StreamNode创建对应的JobVertex，并返回包含该JobVertex配置的StreamConfig
        StreamConfig config =
                currentNodeId.equals(startNodeId)
                        ? createJobVertex(startNodeId, chainInfo)
                        : new StreamConfig(new Configuration());
		
		// 设置每个顶点的基本属性
        setVertexConfig(
                currentNodeId,
                config,
                chainableOutputs,
                nonChainableOutputs,
                chainInfo.getChainedSources());

		// 如果当前currentNodeId与startNodeId一致，证明当前算子链已经完成
        if (currentNodeId.equals(startNodeId)) {
			// 标识StreamNode为当前算子链的起始节点，设置索引位置
            config.setChainStart();
            config.setChainIndex(chainIndex);
config.setOperatorName(streamGraph.getStreamNode(currentNodeId).getOperatorName());
			// 遍历递归栈中返回的出边集合，如果不为空，则证明存在不能被链接在一起的出边，如果为空，则截止目前递归过程中的算子都是被合并链接在一起。
            for (StreamEdge edge : transitiveOutEdges) {
				// 在connect中构建graph
                connect(startNodeId, edge);
            }

			// 设置当前节点的所有出边
            config.setOutEdgesInOrder(transitiveOutEdges);
            // 将chain中所有子节点的StreamConfig写入到headOfChain节点的CHAINED_TASK_CONFIG配置中。
config.setTransitiveChainedTaskConfigs(chainedConfigs.get(startNodeId));

        } else {
			// 如果当前节点是算子链中的子节点
            chainedConfigs.computeIfAbsent(
                    startNodeId, k -> new HashMap<Integer, StreamConfig>());

            config.setChainIndex(chainIndex);
            StreamNode node = streamGraph.getStreamNode(currentNodeId);
            config.setOperatorName(node.getOperatorName());
			// 将当前StreamNode的config记录到以startNodeId开头的算子链config中
            chainedConfigs.get(startNodeId).put(currentNodeId, config);
        }

        config.setOperatorID(currentOperatorId);

		// 如果chainableOutputs为空，证明达到了当前算子链的结尾
        if (chainableOutputs.isEmpty()) {
            config.setChainEnd();
        }

		// 每次递归调用返回的是transitiveOutEdges
        return transitiveOutEdges;

    } else {
        return new ArrayList<>();
    }
}

上述createChain方法是由StreamGraph转换到JobGraph的核心方法。

该方法首先会遍历这个StreamGraph的source节点的所有下游出边，对于每一个边，它将其分类为"可链接（chainable）“和"不可链接（non-chainable）”，分别添加到chainableOutputs和nonChainableOutputs列表中。在具体的实现里，主要逻辑如下：

  1. 遍历输入节点的所有出边，判断出边对应的下游节点与当前上游节点是否可以链接在一起，判断具体逻辑在isChainable()方法中。并根据出边是否可链接，将其添加到不同的集合中。
  2. 接下来，会分两种情况，分别进行递归操作
    • 对于可以chain的情况，会继续递归该下游节点执行createChain方法，以此递归判断整个StreamGraph拓扑结构中，能够chain在一起的所有算子。同时，会将chainIndex+1，chainIndex用来标识算子在算子链中的索引位置。
    • 如果上下游节点不能被chain在一起，则transitiveOutEdges集合中会添加该下游StreamEdge，该集合会在递归调用中返回，上层递归调用根据该变量即可知悉下游递归调用过程中，不能被链接的出边有哪些，后续会调用connect方法来连接。然后会继续调用createChain执行递归，此时当前下游节点即会成为新算子链的起始节点，不过需要注意的是，这里的起始节点索引设置为1，因为只有source节点在算子链中才会被设置为0。
    • 通过上面两种方式的递归，最终递归调用会沿着StreamGraph的拓扑结构一直调用到Sink节点。
  3. 在每个递归执行完成后，也会有两种情况的判断逻辑：
    • 如果currentNodeId与startNodeId一致，说明递归已经回退到了当前算子链的起始节点，当前Chain过程已经完成。此时，会先做一些相关配置，比如标识当前StreamNode为这个算子链的起始节点，设置其在算子链中的索引值，设置操作符名称等作用。并遍历下层递归放回的transitiveOutEdges集合，如果该集合不为空，说明下游节点在递归过程中，存在与当前节点无法链接的节点，为此会调用connect方法进行连接。
    • 如果两者不一致，那么证明当前StreamNode只是这个算子链的一部分，因此，只会设置一些当前节点在算子链中的索引值，操作符名称等信息，并会将当前节点和配置放入以startNodeId为键的chainedConfigsMap对象中，表示当前节点为startNodeId开头的算子链中的一部分。

上面就是这个方法的主要实现逻辑，下面会详细把这个方法展开，重点介绍其中的一些方法实现。

如何判断算子是否可以Chain在一起

两个StreamNode是否可以链接在一起是通过isChainable()方法来判断，更为具体的判断逻辑在该方法的isChainableInput方法中，具体的判断逻辑，如下源码所示：

public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph) {
	// 获取出边对应的下游StreamNode
    StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);

	// 下游Operator的入边只有一个，如果有多个是无法Chain在一起的，因为它还需要接收其他节点的输入
    return downStreamVertex.getInEdges().size() == 1 && 	isChainableInput(edge, streamGraph);
}


private static boolean isChainableInput(StreamEdge edge, StreamGraph streamGraph) {
	// 获取当前StreamEdge对应的上下游StreamNode
    StreamNode upStreamVertex = streamGraph.getSourceVertex(edge);
    StreamNode downStreamVertex = streamGraph.getTargetVertex(edge);

	// 1. 需要对应的slotSharingGroup一样，如果不在同一槽位，意味着仍然需要线程上下文切换，所以无法链接在一起
	// 2. 分区器必须是ForwardPartitioner类型，只有这样才能保证上下游关系一对一
	// 3. 执行模式不能是Batch模式
	// 4. 上下游并发必须一样
	// 5. StreamGraph配置了允许执行算子链接
    if (!(upStreamVertex.isSameSlotSharingGroup(downStreamVertex)
            && areOperatorsChainable(upStreamVertex, downStreamVertex, streamGraph)
            && (edge.getPartitioner() instanceof ForwardPartitioner)
            && edge.getExchangeMode() != StreamExchangeMode.BATCH
            && upStreamVertex.getParallelism() == downStreamVertex.getParallelism()
            && streamGraph.isChainingEnabled())) {

        return false;
    }

    // check that we do not have a union operation, because unions currently only work
    // through the network/byte-channel stack.
    // we check that by testing that each "type" (which means input position) is used only once
    for (StreamEdge inEdge : downStreamVertex.getInEdges()) {
        if (inEdge != edge && inEdge.getTypeNumber() == edge.getTypeNumber()) {
            return false;
        }
    }
    return true;
}

这个方法判断的指标有很多，主要判断逻辑有以下内容：

  1. 需要对应的slotSharingGroup一样，如果不在同一槽位，意味着仍然需要线程上下文切换，所以无法链接在一起
  2. 分区器必须是ForwardPartitioner类型，只有这样才能保证上下游关系一对一
  3. 执行模式不能是Batch模式
  4. 上下游并发必须一样.
  5. StreamGraph配置了允许执行算子链接

基于上述规则判断后，即可判断StreamEdge对应的上下游StreamNode是否可以进行链接。这样的规则也可以指导我们日常中的实际任务开发，比如我们希望算子间能够被链接在一起，提升执行性能的话，就需要让我们的算子间满足上面的规则条件。

创建JobVertex节点

JobVertex 对象的创建是在上面的 createJobVertex() 方法中实现的，这个方法实现比较简单，创建相应的 JobVertex 对象，并把相关的配置信息设置到 JobVertex 对象中就完成了，最终封装在StreamConfig对象中。

connect方法创建JobEdge和IntermediateDataSet对象

上面我们说到，对于不可被链接到同一算子链中的上下游StreamNode是利用connect方法进行连接的，接下来我们就来看看connect方法中执行了什么逻辑，是如何进行算子连接的。

具体的实现源码如下：

private void connect(Integer headOfChain, StreamEdge edge) {

    physicalEdgesInOrder.add(edge);

    Integer downStreamVertexID = edge.getTargetId();
	
	    // 这里 headVertex 指的是 headOfChain 对应的 JobVertex（也是当前 node 对应的 vertex）
    JobVertex headVertex = jobVertices.get(headOfChain);
    JobVertex downStreamVertex = jobVertices.get(downStreamVertexID);

    StreamConfig downStreamConfig = new StreamConfig(downStreamVertex.getConfiguration());

    // 当前下游节点的输入边数+1
downStreamConfig.setNumberOfNetworkInputs(downStreamConfig.getNumberOfNetworkInputs() + 1);

    StreamPartitioner<?> partitioner = edge.getPartitioner();

    ResultPartitionType resultPartitionType;
    switch (edge.getExchangeMode()) {
        case PIPELINED:
            resultPartitionType = ResultPartitionType.PIPELINED_BOUNDED;
            break;
        case BATCH:
            resultPartitionType = ResultPartitionType.BLOCKING;
            break;
        case UNDEFINED:
            resultPartitionType = determineResultPartitionType(partitioner);
            break;
        default:
            throw new UnsupportedOperationException(
                    "Data exchange mode " + edge.getExchangeMode() + " is not supported yet.");
    }

    checkBufferTimeout(resultPartitionType, edge);

    JobEdge jobEdge;
	// 如果当前分区器是点对点的，创建下游节点的上游输入节点和连接
	// 具体的连接逻辑即在connectNewDataSetAsInput方法中
    if (partitioner.isPointwise()) {
        jobEdge =
                downStreamVertex.connectNewDataSetAsInput(
                        headVertex, DistributionPattern.POINTWISE, resultPartitionType);
    } else {
        jobEdge =
                downStreamVertex.connectNewDataSetAsInput(
                        headVertex, DistributionPattern.ALL_TO_ALL, resultPartitionType);
    }
    // set strategy name so that web interface can show it.
    jobEdge.setShipStrategyName(partitioner.toString());
    jobEdge.setDownstreamSubtaskStateMapper(partitioner.getDownstreamSubtaskStateMapper());
    jobEdge.setUpstreamSubtaskStateMapper(partitioner.getUpstreamSubtaskStateMapper());

    if (LOG.isDebugEnabled()) {
        LOG.debug(
                "CONNECTED: {} - {} -> {}",
                partitioner.getClass().getSimpleName(),
                headOfChain,
                downStreamVertexID);
    }
}

真正创建JobEdge执行上下游节点连接的地方在downStreamVertex.connectNewDataSetAsInput方法中，进入到该方法中一探究竟：

public JobEdge connectNewDataSetAsInput(
        JobVertex input, DistributionPattern distPattern, ResultPartitionType partitionType) {
	
	// 创建上游节点的输出中间结果集对象
    IntermediateDataSet dataSet = input.createAndAddResultDataSet(partitionType);

	// 创建对应的JobEdge
    JobEdge edge = new JobEdge(dataSet, this, distPattern);
	// 当前对象(downStreamVertex)的inputs属性添加该JobEdge
    this.inputs.add(edge);
	// 上游节点的输出中间结果集添加下游消费者为该edge
    dataSet.addConsumer(edge);
    return edge;
}

上述的代码清晰展示了连接上下游节点的流程，也对应了我们最开始画的StreamGraph到JobGraph转换的拓扑结构图。

在生成JobGraph时，上游节点并不是直接连接到下游节点的，而是存在一个中间结果集对象，表示上游节点的输出，中间结果集再通过JobEdge连接到下游节点。从而构成了JobGraph结构图中的连接关系。

整体的流程可以再次参考文初的转换图，加深对于整体流程的理解和掌握。

JobGraph 的其他配置

执行完 setChaining() 方法后，下面还有几步操作：

  1. setPhysicalEdges(): 将每个 JobVertex 的入边集合也序列化到该 JobVertex 的 StreamConfig 中 (出边集合已经在 setChaining 的时候写入了)；
  2. setSlotSharingAndCoLocation(): 为每个 JobVertex 指定所属的 SlotSharingGroup 以及设置 CoLocationGroup；
  3. configureCheckpointing(): checkpoint相关的配置；
  4. JobGraphGenerator.addUserArtifactEntries(): 用户依赖的第三方包就是在这里（cacheFile）传给 JobGraph；

在此不再进行赘述。

至此，StreamGraph转换为JobGraph的具体流程就已经梳理完成了，转换流程中需要重点关注的点就是在这一步会进行算子链的优化，以减少算子间的上下文切换开销以及shuffle开销。有兴趣的可以自己翻阅一下源码，这部分内容虽然看着很长，但只要多看几遍，多debug看看具体的执行流程，基本都可以搞明白一二。

参考：
https://matt33.com/2019/12/09/flink-job-graph-3/
https://www.cnblogs.com/GeQian-hq/p/17880647.html
https://wuchong.me/blog/2016/05/10/flink-internals-how-to-build-jobgraph/


http://www.kler.cn/a/463417.html

相关文章：

  • 《Java 数据结构》
  • Java项目实战II基于微信小程序的家庭大厨（开发文档+数据库+源码）
  • 【数据结构】树链刨分
  • SQL 中复杂 CASE WHEN 嵌套逻辑优化
  • 2024年大型语言模型（LLMs）的发展回顾
  • idea 的 springboot项目spring-boot-devtools 自动编译 配置热部署
  • IP寻址映射与网络通信互联
  • [react] 纯组件优化子
  • JMeter脚本参数化与并发策略
  • Vue 针对浏览器参数过长实现浏览器参数加密解密
  • 人工智能机器学习基础篇】——深入详解强化学习 基础知识，理解马尔可夫决策过程（MDP）、策略、价值函数等关键概念
  • FPGA实现HDMI输出
  • 2024树莓派apt源（可用）
  • 【Leetcode 热题 100】79. 单词搜索
  • Amazon Bedrock 实践 - 利用 Llama 3.2 模型分析全球糖尿病趋势
  • uni-app开发-识图小程序-分类识别功能
  • [微服务] - MQ高级
  • 游戏引擎学习第69天
  • doris:基于 Arrow Flight SQL 的高速数据传输链路
  • 小红书ip属地是怎么更新的
  • redis如何实现延时队列
  • 基于单片机的无线智能台灯（论文+源码）
  • STM32单片机芯片与内部54 AT24C02读写 硬件IIC 标准库 HAL库
  • 生态碳汇涡度相关监测与通量数据分析
  • 【Python】selenium结合js模拟鼠标点击、拦截弹窗、鼠标悬停方法汇总（使用 execute_script 执行点击的方法）
  • uniapp——微信小程序，从客户端会话选择文件