说明
当需求变得越来越复杂时，可能在一个 Job 里面算子的个数会达到几十个、甚至上百个，在如此多算子的情况下，整个应用程序就会变得非常复杂，所以在编写 Flink Job 的时候需要知道 Job 的执行计划
Flink 是支持可以获取到整个 Job 的执行计划的
Flink 官网还提供了一个可视化工具 visualizer（可以将执行计划 JSON 绘制出执行图），见 链接
获取执行计划
System.out.println(env.getExecutionPlan());
完整代码
package chongfenqu;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class F11打印执行计划 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> dss = env.fromElements("lilei", "lucy", "lilei", "lili");
SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.map(item -> Tuple2.of(item, 1))
.returns(Types.TUPLE(Types.STRING,Types.INT));
SingleOutputStreamOperator<Tuple2<String, Integer>> sumSOSO = mapSOSO
.keyBy(item -> item.f0)
.sum(1);
sumSOSO.print();
System.out.println(env.getExecutionPlan());
env.execute();
}
}
打印结果
{
"nodes" : [ {
"id" : 1,
"type" : "Source: Collection Source",
"pact" : "Data Source",
"contents" : "Source: Collection Source",
"parallelism" : 1
}, {
"id" : 2,
"type" : "Map",
"pact" : "Operator",
"contents" : "Map",
"parallelism" : 8,
"predecessors" : [ {
"id" : 1,
"ship_strategy" : "REBALANCE",
"side" : "second"
} ]
}, {
"id" : 4,
"type" : "Keyed Aggregation",
"pact" : "Operator",
"contents" : "Keyed Aggregation",
"parallelism" : 8,
"predecessors" : [ {
"id" : 2,
"ship_strategy" : "HASH",
"side" : "second"
} ]
}, {
"id" : 5,
"type" : "Sink: Print to Std. Out",
"pact" : "Data Sink",
"contents" : "Sink: Print to Std. Out",
"parallelism" : 8,
"predecessors" : [ {
"id" : 4,
"ship_strategy" : "FORWARD",
"side" : "second"
} ]
} ]
}
可视化
访问：https://flink.apache.org/visualizer/
将上面的 json 复制到文本框，点击 Draw，如下图；
生成执行计划图
从图中可以看到下面内容呢：
operator name（算子）：比如 source、sink
每个 operator 的并行度：比如 Parallelism: 5
数据下发的类型：比如 FORWARD
参考：
https://blog.csdn.net/weixin_42073629/article/details/120107483