tongchenkeji
发表于：0次点击

请问一下Flink怎么给join设置parallelism?[阿里云]

请问一下Flink怎么给join设置parallelism?

「点点赞赏，手留余香」

    还没有人赞赏，快来当第一个赞赏的人吧！
阿里云# 实时计算 Flink版3179# 流计算2236
=====这是一个广告位，招租中，联系qq 78315851====
阿里云
有人遇到flink打不开web界面吗？[阿里云]
阿里云
有知道在使用flink cdc实现数据同步，如何实现如果服务停止了对数据源表的某个数据进行删除操作？[阿里云]
5 条回复 A 作者 M 管理员

  1. 在Flink中，可以通过设置ExecutionConfig的parallelism属性来控制Join操作的并行度。具体步骤如下：

    1. 创建StreamExecutionEnvironment对象。
    2. 获取StreamExecutionEnvironment对象的getConfig()方法返回的ExecutionConfig对象。
    3. 调用ExecutionConfig对象的setParallelism()方法来设置Join操作的并行度。
    4. 将配置好的ExecutionConfig对象应用到Join操作上。

    示例代码如下：

    // 创建StreamExecutionEnvironment对象StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 获取ExecutionConfig对象ExecutionConfig executionConfig = env.getConfig();// 设置Join操作的并行度为10executionConfig.setParallelism(10);// 将配置好的ExecutionConfig对象应用到Join操作上DataStream<Tuple2<String, Integer>> stream1 = ...; // 第一个数据流DataStream<Tuple2<String, Integer>> stream2 = ...; // 第二个数据流DataStream<Tuple2<String, Integer>> joinedStream = stream1.join(stream2)...; // Join操作
    已喜欢 已反对

  2. 一：如果想通过DataStream对象，需要在提交job时通过StreamExecutionEnvironment#setParallelism(int parallelism)来设置整个job的并行度。
    二：如果想针对一个join操作设置并行度，可以在join操作之前和之后分别设置，但是这种有可能失效，需要实时的观察一下。

    已喜欢 已反对

  3. 可以通过设置 DataStream 的 parallelism 来为 join 操作设置并行度。并行度决定了 Flink 如何分配任务以及使用多少资源来执行这些任务。

    已喜欢 已反对

  4. 在 Apache Flink 中，你可以通过以下步骤给 join 操作设置 parallelism：

    1. 获取 StreamExecutionEnvironment：
      首先，你需要获取 StreamExecutionEnvironment，这是 Flink 作业的执行环境。
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    1. 设置整体并行度：
      如果你想为整个作业设置默认的并行度，可以使用 setParallelism 方法：
    env.setParallelism(parallelism);

    其中 parallelism 是你想要设置的并行度值。

    1. 直接设置 join 算子的并行度：
      对于特定的 join 算子，你可以在定义算子之后直接调用其 setParallelism 方法来设置并行度：
    DataStream<T> joinedStream = stream1.join(stream2)    .where(new KeySelector<T, K> {...})    .equalTo(new KeySelector<T, K> {...})    .window(...)    .apply(new JoinFunction<T, T, R> {...})    .setParallelism(joinParallelism);

    在这个例子中，joinParallelism 是你为 join 算子设置的并行度。

    1. 使用 ExecutionConfig 设置并行度：
      另一种方法是获取 ExecutionConfig 并在其上设置并行度：
    ExecutionConfig executionConfig = env.getConfig();executionConfig.setParallelism(parallelism);

    这将设置所有未显式设置并行度的算子的并行度。

    已喜欢 已反对

  5. 在Flink中，你可以通过以下方法为join操作设置并行度：

    1. 使用ExecutionEnvironment的setParallelism方法。首先，你需要创建一个ExecutionEnvironment实例，然后调用setParallelism方法来设置并行度。例如：
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();env.setParallelism(parallelism);
    1. 使用StreamExecutionEnvironment的createInputFormat方法。这个方法允许你根据输入格式来设置并行度。例如：
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.createInputFormat(MyInputFormat.class, MyPOJO.class).setParallelism(parallelism);
    1. 使用DataStream的assignTimestampsAndWatermarks方法和transform方法。这两个方法都允许你在转换操作中设置并行度。例如：
    DataStream<MyPOJO> dataStream = env.fromElements(...);dataStream.assignTimestampsAndWatermarks(new MyTimestampAssigner()).setParallelism(parallelism);dataStream.transform("Join", TypeInformation.of(MyPOJO.class), new MyJoinFunction()).setParallelism(parallelism);

    注意：这些方法设置的并行度只对当前操作有效，不会影响到其他操作。如果你希望在整个Flink作业中使用相同的并行度，可以在创建ExecutionEnvironment时设置全局并行度。

    已喜欢 已反对

  6. 并行度的设置
    在 Flink 中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。

    （1）代码中设置，我们在代码中，可以很简单地在算子后跟着调用 setParallelism()方法，来设置当前算子的并行度：

    stream.map((_,1)).setParallelism(2)
    这种方式设置的并行度，只针对当前算子有效。另外，我们也可以直接调用执行环境的 setParallelism()方法，全局设定并行度：

    env.setParallelism(2)
    这样代码中所有算子，默认的并行度就都为 2 了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。这里要注意的是，由于 keyBy()方法返回的不是算子，所以无法对 keyBy()设置并行度。

    （2）提交作业时设置

    在使用 flink run 命令提交作业时，可以增加-p 参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

    bin/flink run –p 2 –c com.atguigu.wc.StreamWordCount

    ./FlinkTutorial-1.0-SNAPSHOT.jar
    如果我们直接在 Web UI 上提交作业，也可以在对应输入框中直接添加并行度。

    （3）配置文件中设置

    我们还可以直接在集群的配置文件 flink-conf.yaml 中直接更改默认并行度：

    parallelism.default: 2
    这个设置对于整个集群上提交的所有作业有效，初始值为 1。无论在代码中设置、还是提交时的-p 参数，都不是必须的。所以，在没有指定并行度的时候，就会采用配置文件中的集群默认并行度

    参考：Apache Flink 并行度 Parallelismhttps://blog.csdn.net/lucklilili/article/details/128421426

    已喜欢 已反对