Flink StreamGraph生成过程

🕗 发布于 2024-03-07 11:07 flink 大数据 StreamGraph

文章目录

概要

在 Flink 中，StreamGraph 是数据流的逻辑表示，它描述了如何在 Flink 作业中执行数据流转换。StreamGraph 是 Flink 运行时生成执行计划的基础。
在这里插入图片描述
使用DataStream API开发的应用程序，首先被转换为 Transformation，再被映射为StreamGraph，在客户端进行StreamGraph、JobGraph的转换，提交JobGraph到Flink集群后，Flink集群负责将JobGraph转换为ExecutionGraph，之后进入调度执行阶段。

SteramGraph 核心对象

StreamNode
StreamNode 是 StremGraph 中的节点，从 Transformation 转换而来，可以简单理解为一个 StreamNode 表示一个算子，从逻辑上来说，SteramNode 在 StreamGraph 中存在实体和虚拟的 StreamNode。StremNode 可以有多个输入，也可以有多个输出。
实体的 StreamNode 会最终变成物理算子。虚拟的 StreamNode 会附着在 StreamEdge 上。
StreamEdge
StreamEdge 是 StreamGraph 中的边，用来连接两个 StreamNode,一个 StreamNode 可以有多个出边、入边，StreamEdge 中包含了旁路输出、分区器、字段筛选输出等信息。

SteramGraph 生成过程

StreamGraph 在 FlinkClient 中生成，由 FlinkClient 在提交的时候触发 Flink 应用的 main 方法，用户编写的业务逻辑组装成 Transformation 流水线，在最后调用 StreamExecutionEnvironment.execute() 的时候开始触发 StreamGraph 构建。
StreamGraph在Flink的作业提交前生成，生成StreamGraph的入口在StreamExecutionEnvironment中

    @Internal
    public StreamGraph getStreamGraph() {
        return this.getStreamGraph(this.getJobName());
    }

    @Internal
    public StreamGraph getStreamGraph(String jobName) {
        return this.getStreamGraph(jobName, true);
    }

    @Internal
    public StreamGraph getStreamGraph(String jobName, boolean clearTransformations) {
        StreamGraph streamGraph = this.getStreamGraphGenerator().setJobName(jobName).generate();
        if (clearTransformations) {
            this.transformations.clear();
        }

        return streamGraph;
    }

    private StreamGraphGenerator getStreamGraphGenerator() {
        if (this.transformations.size() <= 0) {
            throw new IllegalStateException("No operators defined in streaming topology. Cannot execute.");
        } else {
            RuntimeExecutionMode executionMode = (RuntimeExecutionMode)this.configuration.get(ExecutionOptions.RUNTIME_MODE);
            return (new StreamGraphGenerator(this.transformations, this.config, this.checkpointCfg, this.getConfiguration())).setRuntimeExecutionMode(executionMode).setStateBackend(this.defaultStateBackend).setChaining(this.isChainingEnabled).setUserArtifacts(this.cacheFile).setTimeCharacteristic(this.timeCharacteristic).setDefaultBufferTimeout(this.bufferTimeout);
        }
    }

StreamGraph实际上是在StreamGraphGenerator中生成的，从SinkTransformation(输出向前追溯到SourceTransformation)。在遍历过程中一边遍历一遍构建StreamGraph，如代码清单所示


@Internal
public class StreamGraphGenerator {
    private final List<Transformation<?>> transformations;
    private StateBackend stateBackend;
    private static final Map<Class<? extends Transformation>, TransformationTranslator<?, ? extends Transformation>> translatorMap;
    protected static Integer iterationIdCounter;
    private StreamGraph streamGraph;
    private Map<Transformation<?>, Collection<Integer>> alreadyTransformed;

    public StreamGraphGenerator(List<Transformation<?>> transformations, ExecutionConfig executionConfig, CheckpointConfig checkpointConfig) {
        this(transformations, executionConfig, checkpointConfig, new Configuration());
    }

    public StreamGraphGenerator(List<Transformation<?>> transformations, ExecutionConfig executionConfig, CheckpointConfig checkpointConfig, ReadableConfig configuration) {
        this.chaining = true;
        this.timeCharacteristic = DEFAULT_TIME_CHARACTERISTIC;
        this.jobName = "Flink Streaming Job";
        this.savepointRestoreSettings = SavepointRestoreSettings.none();
        this.defaultBufferTimeout = -1L;
        this.runtimeExecutionMode = RuntimeExecutionMode.STREAMING;
        this.transformations = (List)Preconditions.checkNotNull(transformations);
        this.executionConfig = (ExecutionConfig)Preconditions.checkNotNull(executionConfig);
        this.checkpointConfig = new CheckpointConfig(checkpointConfig);
        this.configuration = (ReadableConfig)Preconditions.checkNotNull(configuration);
    }

    public StreamGraph generate() {
        this.streamGraph = new StreamGraph(this.executionConfig, this.checkpointConfig, this.savepointRestoreSettings);
        this.shouldExecuteInBatchMode = this.shouldExecuteInBatchMode(this.runtimeExecutionMode);
        this.configureStreamGraph(this.streamGraph);
        this.alreadyTransformed = new HashMap();
        Iterator var1 = this.transformations.iterator();

        while(var1.hasNext()) {
            Transformation<?> transformation = (Transformation)var1.next();
            this.transform(transformation);
        }

        StreamGraph builtStreamGraph = this.streamGraph;
        this.alreadyTransformed.clear();
        this.alreadyTransformed = null;
        this.streamGraph = null;
        return builtStreamGraph;
    }
 
    private Collection<Integer> transform(Transformation<?> transform) {
        if (this.alreadyTransformed.containsKey(transform)) {
            return (Collection)this.alreadyTransformed.get(transform);
        } else {
            LOG.debug("Transforming " + transform);
            if (transform.getMaxParallelism() <= 0) {
                int globalMaxParallelismFromConfig = this.executionConfig.getMaxParallelism();
                if (globalMaxParallelismFromConfig > 0) {
                    transform.setMaxParallelism(globalMaxParallelismFromConfig);
                }
            }

            transform.getOutputType();
            TransformationTranslator<?, Transformation<?>> translator = (TransformationTranslator)translatorMap.get(transform.getClass());
            Collection transformedIds;
            if (translator != null) {
                transformedIds = this.translate(translator, transform);
            } else {
                transformedIds = this.legacyTransform(transform);
            }

            if (!this.alreadyTransformed.containsKey(transform)) {
                this.alreadyTransformed.put(transform, transformedIds);
            }

            return transformedIds;
        }
    }

    private Collection<Integer> legacyTransform(Transformation<?> transform) {
        Collection transformedIds;
        if (transform instanceof FeedbackTransformation) {
            transformedIds = this.transformFeedback((FeedbackTransformation)transform);
        } else {
            if (!(transform instanceof CoFeedbackTransformation)) {
                throw new IllegalStateException("Unknown transformation: " + transform);
            }

            transformedIds = this.transformCoFeedback((CoFeedbackTransformation)transform);
        }

        if (transform.getBufferTimeout() >= 0L) {
            this.streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());
        } else {
            this.streamGraph.setBufferTimeout(transform.getId(), this.defaultBufferTimeout);
        }

        if (transform.getUid() != null) {
            this.streamGraph.setTransformationUID(transform.getId(), transform.getUid());
        }

        if (transform.getUserProvidedNodeHash() != null) {
            this.streamGraph.setTransformationUserHash(transform.getId(), transform.getUserProvidedNodeHash());
        }

        if (!this.streamGraph.getExecutionConfig().hasAutoGeneratedUIDsEnabled() && transform instanceof PhysicalTransformation && transform.getUserProvidedNodeHash() == null && transform.getUid() == null) {
            throw new IllegalStateException("Auto generated UIDs have been disabled but no UID or hash has been assigned to operator " + transform.getName());
        } else {
            if (transform.getMinResources() != null && transform.getPreferredResources() != null) {
                this.streamGraph.setResources(transform.getId(), transform.getMinResources(), transform.getPreferredResources());
            }

            this.streamGraph.setManagedMemoryUseCaseWeights(transform.getId(), transform.getManagedMemoryOperatorScopeUseCaseWeights(), transform.getManagedMemorySlotScopeUseCases());
            return transformedIds;
        }
    }

    private <T> Collection<Integer> transformFeedback(FeedbackTransformation<T> iterate) {
        if (this.shouldExecuteInBatchMode) {
            throw new UnsupportedOperationException("Iterations are not supported in BATCH execution mode. If you want to execute such a pipeline, please set the '" + ExecutionOptions.RUNTIME_MODE.key() + "'=" + RuntimeExecutionMode.STREAMING.name());
        } else if (iterate.getFeedbackEdges().size() <= 0) {
            throw new IllegalStateException("Iteration " + iterate + " does not have any feedback edges.");
        } else {
            List<Transformation<?>> inputs = iterate.getInputs();
            Preconditions.checkState(inputs.size() == 1);
            Transformation<?> input = (Transformation)inputs.get(0);
            List<Integer> resultIds = new ArrayList();
            Collection<Integer> inputIds = this.transform(input);
            resultIds.addAll(inputIds);
            if (this.alreadyTransformed.containsKey(iterate)) {
                return (Collection)this.alreadyTransformed.get(iterate);
            } else {
                Tuple2<StreamNode, StreamNode> itSourceAndSink = this.streamGraph.createIterationSourceAndSink(iterate.getId(), getNewIterationNodeId(), getNewIterationNodeId(), iterate.getWaitTime(), iterate.getParallelism(), iterate.getMaxParallelism(), iterate.getMinResources(), iterate.getPreferredResources());
                StreamNode itSource = (StreamNode)itSourceAndSink.f0;
                StreamNode itSink = (StreamNode)itSourceAndSink.f1;
                this.streamGraph.setSerializers(itSource.getId(), (TypeSerializer)null, (TypeSerializer)null, iterate.getOutputType().createSerializer(this.executionConfig));
                this.streamGraph.setSerializers(itSink.getId(), iterate.getOutputType().createSerializer(this.executionConfig), (TypeSerializer)null, (TypeSerializer)null);
                resultIds.add(itSource.getId());
                this.alreadyTransformed.put(iterate, resultIds);
                List<Integer> allFeedbackIds = new ArrayList();
                Iterator var10 = iterate.getFeedbackEdges().iterator();

                while(var10.hasNext()) {
                    Transformation<T> feedbackEdge = (Transformation)var10.next();
                    Collection<Integer> feedbackIds = this.transform(feedbackEdge);
                    allFeedbackIds.addAll(feedbackIds);
                    Iterator var13 = feedbackIds.iterator();

                    while(var13.hasNext()) {
                        Integer feedbackId = (Integer)var13.next();
                        this.streamGraph.addEdge(feedbackId, itSink.getId(), 0);
                    }
                }

                String slotSharingGroup = this.determineSlotSharingGroup((String)null, allFeedbackIds);
                if (slotSharingGroup == null) {
                    slotSharingGroup = "SlotSharingGroup-" + iterate.getId();
                }

                itSink.setSlotSharingGroup(slotSharingGroup);
                itSource.setSlotSharingGroup(slotSharingGroup);
                return resultIds;
            }
        }
    }

    private <F> Collection<Integer> transformCoFeedback(CoFeedbackTransformation<F> coIterate) {
        if (this.shouldExecuteInBatchMode) {
            throw new UnsupportedOperationException("Iterations are not supported in BATCH execution mode. If you want to execute such a pipeline, please set the '" + ExecutionOptions.RUNTIME_MODE.key() + "'=" + RuntimeExecutionMode.STREAMING.name());
        } else {
            Tuple2<StreamNode, StreamNode> itSourceAndSink = this.streamGraph.createIterationSourceAndSink(coIterate.getId(), getNewIterationNodeId(), getNewIterationNodeId(), coIterate.getWaitTime(), coIterate.getParallelism(), coIterate.getMaxParallelism(), coIterate.getMinResources(), coIterate.getPreferredResources());
            StreamNode itSource = (StreamNode)itSourceAndSink.f0;
            StreamNode itSink = (StreamNode)itSourceAndSink.f1;
            this.streamGraph.setSerializers(itSource.getId(), (TypeSerializer)null, (TypeSerializer)null, coIterate.getOutputType().createSerializer(this.executionConfig));
            this.streamGraph.setSerializers(itSink.getId(), coIterate.getOutputType().createSerializer(this.executionConfig), (TypeSerializer)null, (TypeSerializer)null);
            Collection<Integer> resultIds = Collections.singleton(itSource.getId());
            this.alreadyTransformed.put(coIterate, resultIds);
            List<Integer> allFeedbackIds = new ArrayList();
            Iterator var7 = coIterate.getFeedbackEdges().iterator();

            while(var7.hasNext()) {
                Transformation<F> feedbackEdge = (Transformation)var7.next();
                Collection<Integer> feedbackIds = this.transform(feedbackEdge);
                allFeedbackIds.addAll(feedbackIds);
                Iterator var10 = feedbackIds.iterator();

                while(var10.hasNext()) {
                    Integer feedbackId = (Integer)var10.next();
                    this.streamGraph.addEdge(feedbackId, itSink.getId(), 0);
                }
            }

            String slotSharingGroup = this.determineSlotSharingGroup((String)null, allFeedbackIds);
            itSink.setSlotSharingGroup(slotSharingGroup);
            itSource.setSlotSharingGroup(slotSharingGroup);
            return Collections.singleton(itSource.getId());
        }
    }

    private Collection<Integer> translate(TransformationTranslator<?, Transformation<?>> translator, Transformation<?> transform) {
        Preconditions.checkNotNull(translator);
        Preconditions.checkNotNull(transform);
        List<Collection<Integer>> allInputIds = this.getParentInputIds(transform.getInputs());
        if (this.alreadyTransformed.containsKey(transform)) {
            return (Collection)this.alreadyTransformed.get(transform);
        } else {
            String slotSharingGroup = this.determineSlotSharingGroup(transform.getSlotSharingGroup(), (Collection)allInputIds.stream().flatMap(Collection::stream).collect(Collectors.toList()));
            Context context = new StreamGraphGenerator.ContextImpl(this, this.streamGraph, slotSharingGroup, this.configuration);
            return this.shouldExecuteInBatchMode ? translator.translateForBatch(transform, context) : translator.translateForStreaming(transform, context);
        }
    }

    private List<Collection<Integer>> getParentInputIds(@Nullable Collection<Transformation<?>> parentTransformations) {
        List<Collection<Integer>> allInputIds = new ArrayList();
        if (parentTransformations == null) {
            return allInputIds;
        } else {
            Iterator var3 = parentTransformations.iterator();

            while(var3.hasNext()) {
                Transformation<?> transformation = (Transformation)var3.next();
                allInputIds.add(this.transform(transformation));
            }

            return allInputIds;
        }
    }

    private String determineSlotSharingGroup(String specifiedGroup, Collection<Integer> inputIds) {
        if (specifiedGroup != null) {
            return specifiedGroup;
        } else {
            String inputGroup = null;
            Iterator var4 = inputIds.iterator();

            while(var4.hasNext()) {
                int id = (Integer)var4.next();
                String inputGroupCandidate = this.streamGraph.getSlotSharingGroup(id);
                if (inputGroup == null) {
                    inputGroup = inputGroupCandidate;
                } else if (!inputGroup.equals(inputGroupCandidate)) {
                    return "default";
                }
            }

            return inputGroup == null ? "default" : inputGroup;
        }
    }

    static {
        DEFAULT_TIME_CHARACTERISTIC = TimeCharacteristic.ProcessingTime;
        Map<Class<? extends Transformation>, TransformationTranslator<?, ? extends Transformation>> tmp = new HashMap();
        tmp.put(OneInputTransformation.class, new OneInputTransformationTranslator());
        tmp.put(TwoInputTransformation.class, new TwoInputTransformationTranslator());
        tmp.put(MultipleInputTransformation.class, new MultiInputTransformationTranslator());
        tmp.put(KeyedMultipleInputTransformation.class, new MultiInputTransformationTranslator());
        tmp.put(SourceTransformation.class, new SourceTransformationTranslator());
        tmp.put(SinkTransformation.class, new SinkTransformationTranslator());
        tmp.put(LegacySinkTransformation.class, new LegacySinkTransformationTranslator());
        tmp.put(LegacySourceTransformation.class, new LegacySourceTransformationTranslator());
        tmp.put(UnionTransformation.class, new UnionTransformationTranslator());
        tmp.put(PartitionTransformation.class, new PartitionTransformationTranslator());
        tmp.put(SideOutputTransformation.class, new SideOutputTransformationTranslator());
        tmp.put(ReduceTransformation.class, new ReduceTransformationTranslator());
        tmp.put(TimestampsAndWatermarksTransformation.class, new TimestampsAndWatermarksTransformationTranslator());
        tmp.put(BroadcastStateTransformation.class, new BroadcastStateTransformationTranslator());
        tmp.put(KeyedBroadcastStateTransformation.class, new KeyedBroadcastStateTransformationTranslator());
        translatorMap = Collections.unmodifiableMap(tmp);
        iterationIdCounter = 0;
    } 
}

原文地址：https://blog.csdn.net/gwc791224/article/details/136437944

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：如何开发一个高质量的公共组件
下一篇：【教程】uni-app iOS打包解决profile文件与私钥证书不匹配问题

leetcode289:生命游戏
根据，简称为，是英国数学家约翰·何顿·康威在 1970 年发明的细胞自动机。给定一个包含m × n个格子的面板，每一个格子都可以看成是一个细胞。每个细胞都具有一个初始状态：1即为（live），或0即为
阅读更多2024-10-20
MongoDB数据恢复
注意：两个MongoDB的版本要一致，本文使用的是mongo:4.2.24。先把K8S上面的MongoDB 容器停止（可以把副本改成0）。1、将容器挂载MongoDB的数据目录备份到本地。经常是数据文
阅读更多2024-10-20
C#中实现事务
C#中实现事务
阅读更多2024-10-20
【LeetCode每日一题】——560.和为 K 的子数组
给你一个整数数组 nums 和一个整数 k ，请你统计并返回该数组中和为 k 的子数组的个数。子数组是数组中元素的连续非空序列。
阅读更多2024-10-20
「漏洞复现」满客宝智慧食堂系统 selectUserByOrgId 未授权访问漏洞
请勿利用文章内的相关技术从事非法测试，由于传播、利用此文所提供的信息而造成的任何直接或者间接的后果及损失，均由使用者本人负责，作者不为此承担任何责任。工具来自网络，安全性自测，如有侵权请联系删除。本次
阅读更多2024-10-20
React面试题目（从基本到高级）
React前端面试常见题目涵盖了React的基础概念、组件、状态管理、生命周期、性能优化等多个方面。
阅读更多2024-10-20
12.个人博客系统（Java项目基于spring和vue）
1 在校学习的学生，可用于日常学习使用或是毕业设计使用 2 毕业一到两年的开发人员，用于锻炼自己的独立功能模块设计能力，增强代码编写能力。 3 亦可以部署为商化项目使用。 4 需要完整资料及源码
阅读更多2024-10-20
YoloV8改进策略：注意力改进|DeBiFormer，可变形双级路由注意力|引入DeBiLevelRoutingAttention注意力模块（全网首发）
本次改进的核心在于将DeBiLevelRoutingAttention模块嵌入到YoloV8的主干网络中，具体位置是在SPPF（Spatial Pyramid Pooling Fast）模块之后。这一
阅读更多2024-10-20
word取消自动单词首字母大写
情况说明：在word输入单词后首字母会自动变成大写取消单词首字母大写步骤：（1）点击菜单栏文件（2）点击“更多”——>“选项”（3）点击“校对”——>“自动更正选项”（4）取消“句首字母大
阅读更多2024-10-20
web前端网页用户注册页面
【代码】web前端网页用户注册页面。
阅读更多2024-10-20

Flink StreamGraph生成过程

文章目录

概要

SteramGraph 核心对象

SteramGraph 生成过程

相关文章