以批處理程序中的wordCount為例，調(diào)試一下Plan的生成以及對于優(yōu)化計劃OptimizedPlan的轉(zhuǎn)化過程。至于如何進入Execute接口的上述過程將要 放到以后進行說明，直接從生成批處理計劃Plan開始：
我們可以見到當用戶使用env.ecxecute（）的時候，進入Localexecutor類生成Plan，Plan p來接收createProgramPlan的返回值，傳遞的參數(shù)僅為jobName，jobName一般是創(chuàng)建這個作業(yè)的日期和時間。下面我們就進入這個方法createProargamoPlan中?？梢栽趀xecutionEnvironment類中找到兩個名字相同的方法createProgramPlan，這是java函數(shù)的重載。通過jobName進入第一個，第一個返回第二個方法的bool值，默認為真，代表重建一個作業(yè)的sink執(zhí)行，即一個新階段的作業(yè)的執(zhí)行：
同名類下的第二個創(chuàng)建方法
可以看到算子翻譯器OperatorTranslation（）對批處理程序中的算子進行翻譯轉(zhuǎn)換。跳轉(zhuǎn)進入到這個方法中可以發(fā)現(xiàn)通過從sink節(jié)點開始遍歷，將遇到的算子轉(zhuǎn)換成Plan的形式，使用一個translate的方法。translate負責將flink中的算子獲取并且注冊。如何實現(xiàn)算子的翻譯以及注冊為Operator類型可以見下面的博客，該博客介紹了從批處理程序生成優(yōu)化計劃的過程：
https://blog.csdn.net/yanghua_kobe/article/details/55224512?locationNum=5&fps=1
對應源代碼，生成Plan主是通過下面的這個語句：
那么是怎么進入到計劃優(yōu)化，即PanOptimizer class里面的呢？還是通過LocalExecutor中的execute方法生成的，返回類型為jobExecutionResult。關(guān)鍵的兩句如下：
通過第一句可以進入到Optimizer類中，這個Optimizer類它接受用戶指定的程序計劃并創(chuàng)建一個優(yōu)化計劃，其中包含有關(guān)如何進行物理執(zhí)行的準確描述。 它首先將用戶程序轉(zhuǎn)換為內(nèi)部優(yōu)化程序表示，然后在運輸策略和本地策略的不同備選方案之間進行選擇。進入到Optimizer類下面后可以看到兩個名為compile的方法，返回類
型均為OptimizedPlan。話說flink中很喜歡如下操作，首先當用戶未給出另一形參的時候先通過一個方法從第一個參數(shù)中得到另一個類型的返回值，在通過return調(diào)用重載函數(shù)。上圖中的OptimizedPostPass得到以后，調(diào)用下面的compile方法。
此步驟執(zhí)行以下操作：
1）它為每個運算符創(chuàng)建一個優(yōu)化器計劃節(jié)點
2）它通過通道連接它們
3）它查找有關(guān)本地策略和頻道類型的提示相應地設置類型和策略
4）它估算數(shù)據(jù)源的數(shù)據(jù)量和通過計劃傳播這些估算值
首先確定dataSink節(jié)點，若有多個就通過上述遍歷的方式確定一個root節(jié)點。之后通過BranchVisitor進行分支定界，估算PlanNode的成本，具體的成本估算方式我之后會給出介紹。首先跳過以上步驟，optimizer包下的costs包可以進行成本估算。內(nèi)涵默認的成本估算以及根據(jù)PlanNode之間的數(shù)據(jù)傳輸策略進行估算，選取最優(yōu)的執(zhí)行計劃作為bestPlan。通過bestPlan的隊首節(jié)點可以找到sink或者sink節(jié)點的集合。根據(jù)這個bestPlanRoot進行下面的操作：
PlanFinalizer類通過計算每個PlanNode的代價、權(quán)重之后，返回Optimizer類型的結(jié)果。
OptimizedPlan主要封裝了如下這些屬性：
dataSources：SourcePlanNode集合；
dataSinks：SinkPlanNode集合；
allNodes：優(yōu)化后計劃中的所有PlanNode節(jié)點集合；
originalProgram：最初未被優(yōu)化的Plan對象
