Apache Flink:setParallelism（）和setMaxParallelism（）的区别是什么

Question

我尝试为Flink作业设置最大并行度，使用ExecutionConfig.setMaxParallelism（）方法，但它似乎不起作用。

我还修改了标准WordCount示例以运行一些测试，似乎setMaxParallelism（）方法对本地环境或独立集群没有任何影响。

setMaxParallelism（）如何工作？

Answer 1

今天我又运行了一些测试，使用流而不是数据集。这次我看到了setMaxParallelism（）的效果。

    public static void main(String[] args) throws Exception
    {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.getConfig().setMaxParallelism(4); // <-- effect

        DataStream<String> text = env.fromElements(WORDS);

        DataStream<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer()).keyBy(0).sum(1);

        counts.writeAsCsv("test.dat");

        env.execute("WordCount Example");
    }

客户看到的有趣错误，

Caused by: org.apache.flink.runtime.JobException: Vertex Flat Map's parallelism (8) is higher than the max parallelism (4). Please lower the parallelism or increase the max parallelism.
        at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:188)
        at org.apache.flink.runtime.executiongraph.ExecutionGraph.attachJobGraph(ExecutionGraph.java:830)
        at org.apache.flink.runtime.executiongraph.ExecutionGraphBuilder.buildGraph(ExecutionGraphBuilder.java:232)
        at org.apache.flink.runtime.executiongraph.ExecutionGraphBuilder.buildGraph(ExecutionGraphBuilder.java:100)
        at org.apache.flink.runtime.jobmaster.JobMaster.createExecutionGraph(JobMaster.java:1152)
        at org.apache.flink.runtime.jobmaster.JobMaster.createAndRestoreExecutionGraph(JobMaster.java:1132)
        at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:294)
        at org.apache.flink.runtime.jobmaster.JobManagerRunner.<init>(JobManagerRunner.java:157)
        ... 10 more

谢谢

Answer 2

Flink提供两种设置：

• 将作业或操作员的并行度设置为x，即操作员的并行任务数
• setMaxParallelism（y）控制键控状态可以分配到的最大任务数，即操作符的最大有效并行度。操作员仍然可以有更多的任务，但其中只有y个任务将分配键控状态并可用于处理。分发密钥状态的单元称为密钥组

留档将更详细地解释这些概念。