flink 中窗口后的 reduce 并行数为什么是1

flink 中窗口后的 reduce 并行数为什么是1

如何控制并行数，用一个变量存起来呗，parallelism 变量存放某个 *** 作的并行数。一般是有默认配置或者用户指定。
那为什么窗口后reduce 只有一个并行数。且无法改变。
因为强制把此 *** 作的并行数设置为1 了呗。

// 这是 AllWindowedStream类的 reduce(...) 中的一段。很明显在强制 *** 作。
 return input.transform(opName, resultType, operator).forceNonParallel();
 
// 这是 SingleOutputStreamOperator类中的方法，就是在设置并行数
public SingleOutputStreamOperator forceNonParallel() {
        transformation.setParallelism(1);
        transformation.setMaxParallelism(1);
        nonParallel = true;
        return this;
    }
 

是不是不能改啊？对的

// 这里是SingleOutputStreamOperator类中方法。
//因为强制设置，这个返回false;
    private boolean canBeParallel() {
         return !nonParallel;
    }
// 设置并行数
  public SingleOutputStreamOperator setParallelism(int parallelism) {
       //这里会check。
        OperatorValidationUtils.validateParallelism(parallelism, canBeParallel());
        transformation.setParallelism(parallelism);

        return this;
    }    
 // 工具类   
   public static void validateParallelism(int parallelism, boolean canBeParallel) {
        Preconditions.checkArgument(
        //canBeParallel为false，如果parallelism不等于1，则传入false。
                canBeParallel || parallelism == 1,
                "The parallelism of non parallel operator must be 1.");
        Preconditions.checkArgument(
                parallelism > 0 || parallelism == ExecutionConfig.PARALLELISM_DEFAULT,
                "The parallelism of an operator must be at least 1, or ExecutionConfig.PARALLELISM_DEFAULT (use system default).");
    } 
     
      public static void checkArgument(boolean condition, @Nullable Object errorMessage) {
      // 如果条件为false，将报错。
        if (!condition) {
            throw new IllegalArgumentException(String.valueOf(errorMessage));
        }
    }

好了，一个简单的源码。