作者：京東物流 吳云濤

前言

提交一個DataSteam 的 Flink應用，需要經過 StreamGraph、JobGraph、ExecutionGraph 三個階段的轉換生成可成執行的有向無環圖（DAG），并在 Flink 集群上運行。而提交一個 Flink SQL 應用，其執行流程也類似，只是多了一步使用 flink-table-planer 模塊從SQL轉換成 StreamGraph 的過程。以下是利用Flink的 StreamGraph 通過低代碼的方式，來實現StreamGraph的生成，并最終實現 Flink 程序零代碼開發的解決方案。

一、Flink 相關概念

在Flink程序中，每個算子被稱作Operator，通過各個算子的處理最終得到期望的加工后數據。比如下面這段程序中，增加了Source, Fiter, Map, Sink 4個算子。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream dataStream = env.addSource(new FlinkKafkaConsumer("topic"));

DataStream filteredStream = dataStream.filter(new FilterFunction() {
    @Override
    public boolean filter(Object value) throws Exception {return true;}
});

DataStream mapedStream = filteredStream.map(new MapFunction() {
    @Override
    public Object map(Object value) throws Exception {return value;}
});

mapedStream.addSink(new DiscardingSink());
env.execute("test-job");

StreamGraph

Flink的邏輯執行圖，描述了整個流處理任務的流程和數據流轉遞規則，包括了數據源（Source）、轉換算子(Transform)、數據目的端(Sink)等元素，以及它們之間的依賴關系和傳輸規則。StreamGraph是通過Flink的API或者DSL來構建的向無環圖（DAG），它與JobGraph之間是一一對應的關系。StreamGraph中的頂點稱為streamNode，是用來表示Operator算子的類，包含了算子uid、并行度，是否共享slot（SlotSharingGroup）等信息。邊稱作streamEdge。通過StreamingJobGraphGenerator類生成JobGraph。

JobGraph

StreamGraph 經過 flink-optimizer 模塊優化后生成 JobGraph。生成 JobGraph 時，會將多個滿足條件的算子chain 鏈接到一起作為一個頂點（JobVertex), 在運行時對應1個 Task。Task 是 Flink 程序的基本執行單元，任務調度時將Task分配到TaskManager上執行。

ExecutionGraph

物理執行圖是由JobGraph轉換而來，描述了整個流處理任務的物理執行細節，包括了任務的調度、任務的執行順序、任務之間的數據傳輸、任務的狀態管理等。Task會在步驟中拆分為多個SubTask。對應Task中的每個并行度。

Physical Graph

PhysicalGraph是在執行時的ExecutionGraph。ExecutionGraph中的每一個頂點ExecutionJobVertex都對應一個或多個頂點ExecutionVertex，它們是物理執行圖中的節點。

二、畫布模式實現思路

實現流程

首先，我們采用畫布模式（拖拉拽方式）來實現Flink程序的組裝，將極大程度上方便我們復用部分加工的算子，最終實現零代碼的Flink應用開發。我們通過繪圖的方式，直接將內置的算子繪制在圖標上。如下所示：

構建有向無環圖（DAG），并持久化。通過拖拉拽的方式（畫布模式）構建你的Flink應用，后端的持久化存儲采用鄰接表方式。我們在 mysql 關系數據庫中將 Node(算子：Source、Sink、中間加工邏輯算子)存儲到 flink_node 表中；將邊存到一張 flink_realation 表中。

重新組將Flink作業
要組裝以上畫布模式的Flink應用，首先需要初始化好 StreamExecutionEnvironment 相關參數，其次將上述表中的 flink_node 和flink_edge 轉化為DataStream，并將轉化出的 DataStream 合理地拼接成一個 DataStream API Flink 應用程序。
在將flink_node、flink_edge轉為為DataStream時選擇何種遍歷算法來組裝呢？我們知道有向無環圖的遍歷最常用的有：深度優先遍歷（DFS）和廣度優先遍歷（BFS）。這里我們采用了BFS算法+層序遍歷的方式，BFS便于在組裝的過程中將已visit到的node節點拼裝到其parent 的節點上。

總結

在實際的實現過程中，遇到的問題往往比以上復雜很多。比如需要將更多的信息存儲在node節點和edge邊上。node上需要存儲并行度、算子處理前后的表schema等；edge需要存儲keyby的字段、上下游之間的數據shuffle的方式等等。此外在內置的算子無法滿足用戶需求時，還需要考慮如何友好的支持自定義算子(UDF)的嵌入等問題。

審核編輯 黃宇