為什么80%的碼農(nóng)都做不了架構(gòu)師？>>> ??

序

本文主要研究一下flink的Execution Plan Visualization

實例

代碼

@Testpublic void testExecutionPlan(){final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String,Integer>> dataStream = env.fromElements(WORDS).flatMap(new WordCountTest.Tokenizer()).keyBy(0).sum(1);dataStream.print();System.out.println(env.getExecutionPlan());}

json

{"nodes": [{"id": 1,"type": "Source: Collection Source","pact": "Data Source","contents": "Source: Collection Source","parallelism": 1},{"id": 2,"type": "Flat Map","pact": "Operator","contents": "Flat Map","parallelism": 4,"predecessors": [{"id": 1,"ship_strategy": "REBALANCE","side": "second"}]},{"id": 4,"type": "Keyed Aggregation","pact": "Operator","contents": "Keyed Aggregation","parallelism": 4,"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": 4,"predecessors": [{"id": 4,"ship_strategy": "FORWARD","side": "second"}]}] }

可視化

打開flink plan visualizer將上面的json，輸入到文本框，點擊Draw進(jìn)行可視化如下：

StreamExecutionEnvironment.getExecutionPlan

flink-streaming-java_2.11-1.7.1-sources.jar!/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java

@Public public abstract class StreamExecutionEnvironment {//....../*** Creates the plan with which the system will execute the program, and* returns it as a String using a JSON representation of the execution data* flow graph. Note that this needs to be called, before the plan is* executed.** @return The execution plan of the program, as a JSON String.*/public String getExecutionPlan() {return getStreamGraph().getStreamingPlanAsJSON();}/*** Getter of the {@link org.apache.flink.streaming.api.graph.StreamGraph} of the streaming job.** @return The streamgraph representing the transformations*/@Internalpublic StreamGraph getStreamGraph() {if (transformations.size() <= 0) {throw new IllegalStateException("No operators defined in streaming topology. Cannot execute.");}return StreamGraphGenerator.generate(this, transformations);}//...... }

• StreamExecutionEnvironment的getExecutionPlan方法調(diào)用了getStreamGraph方法；getStreamGraph方法使用StreamGraphGenerator.generate生成了StreamGraph；之后就是調(diào)用StreamGraph.getStreamingPlanAsJSON來獲取json格式的execution plan

StreamGraph.getStreamingPlanAsJSON

flink-streaming-java_2.11-1.7.1-sources.jar!/org/apache/flink/streaming/api/graph/StreamGraph.java

@Internal public class StreamGraph extends StreamingPlan {private static final Logger LOG = LoggerFactory.getLogger(StreamGraph.class);private String jobName = StreamExecutionEnvironment.DEFAULT_JOB_NAME;private final StreamExecutionEnvironment environment;private final ExecutionConfig executionConfig;private final CheckpointConfig checkpointConfig;private boolean chaining;private Map<Integer, StreamNode> streamNodes;private Set<Integer> sources;private Set<Integer> sinks;private Map<Integer, Tuple2<Integer, List<String>>> virtualSelectNodes;private Map<Integer, Tuple2<Integer, OutputTag>> virtualSideOutputNodes;private Map<Integer, Tuple2<Integer, StreamPartitioner<?>>> virtualPartitionNodes;protected Map<Integer, String> vertexIDtoBrokerID;protected Map<Integer, Long> vertexIDtoLoopTimeout;private StateBackend stateBackend;private Set<Tuple2<StreamNode, StreamNode>> iterationSourceSinkPairs;//......public String getStreamingPlanAsJSON() {try {return new JSONGenerator(this).getJSON();}catch (Exception e) {throw new RuntimeException("JSON plan creation failed", e);}}//...... }

• StreamGraph的getStreamingPlanAsJSON方法使用JSONGenerator來序列化自己，返回json格式的execution plan

小結(jié)

• flink提供了flink plan visualizer的在線地址，用于進(jìn)行execution plan的可視化，它接收json形式的execution plan
• StreamExecutionEnvironment的getExecutionPlan方法調(diào)用了getStreamGraph方法；getStreamGraph方法使用StreamGraphGenerator.generate生成了StreamGraph
• StreamGraph的getStreamingPlanAsJSON方法使用JSONGenerator來序列化自己，返回json格式的execution plan

doc

• Execution Plans
• flink plan visualizer

轉(zhuǎn)載于:https://my.oschina.net/go4it/blog/3009595

總結(jié)

以上是生活随笔為你收集整理的聊聊flink的Execution Plan Visualization的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。