如何在纯 java 中设置 flink 并行度（IDEA）

Question

我正在使用以下 scala 代码来 运行 我的 flink 流式传输作业

val mystream = StreamExecutionEnvironment.getExecutionEnvironment
    mystream.addSource(new mySource(params))
      .map(new myMap(params))
      .addSink(new mySink(params)).setParallelism(1)
    mystream.setParallelism(1)
    mystream.execute("My Streaming")

当我使用flink run -p 1时，并行度为1（不知道-p 是否有效或代码是否有效）。当我使用纯 java 到 运行 时，（在 IDEA 中我认为它在纯 java 中是 运行），并行度通常为 5，这表明我的代码不起作用.如何控制？

---

正如最佳答案所建议的那样，以下代码也不起作用，仍然具有 5 的并行性。

val mystream = StreamExecutionEnvironment.getExecutionEnvironment
    mystream.addSource(new mySource(params))
      .map(new myMap(params))
      .addSink(new mySink(params))
    mystream.setParallelism(1)
    mystream.execute("My Streaming")

Answer 1

您设置了环境的默认并行度。

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
env.addSource(...)

使用 .addSink(new mySink(params)).setParallelism(1) 覆盖特定运算符的默认并行度。