[[391967]]

這次從一張圖說起，就是下面這幅圖，當我畫完他的時候，我就感覺無比的舒爽。讓整個執行流程，可控，并且可擴展。還記得當初上學的時候，數學建模課上，老師為了“求證一張四條腿的椅子，四條腿處在一個平面上的概率”，寫了滿滿四黑板板書，然后心滿意足的，跟一臉蒙的我們說，“看，多美啊!”。我現在有點理解他當時的狀態了。

有眼尖的同學，可能已經發現了，沒錯，這里借鑒了Flink的流程設計。

簡單的說，輸入-> 轉換 -> 輸出 經典的三段論式構造。

這里我們把輸入 SourceFunction 和 輸出 SinkFunction 進行了抽象，但是并沒有像Flink那樣，輸入輸出集成相同的接口，可以自由轉換，從需求上來看，是沒有必要Sql -to- Sql -to- Sql 的必要的。

在轉換過程中，我們引用了DSL，來承接前面的SQL解析，以及后續的SQL生成，于是，自然的將轉換分成了2段，即輸入轉換(SourceMapper) 和 輸出轉換(SinkMapper)

而定義轉換與否，以及轉換動如何完成，我們可以輕松的通過java自帶的 Function 接口來實現。

一個輸入是可以接對接N條輸出的，輸出的方向也完全由實現 SinkFunction 的對象來自實現。

如果后續，把任務并行化，Pipline 對外提供高階函數的話， 畫美不看...

DEMO：

public static void main(String[] args) throws IOException { 
      //輸入 
      FileSourceFunction fileSourceFunction = new FileSourceFunction(); 
      fileSourceFunction.init(new File("fff.txt")); 
 
      //輸出 
      PrintlnSinkFunction<PrintStream> printSinkFunction = new PrintlnSinkFunction(); 
      ListSinkFunction listSinkFunction = new ListSinkFunction(); 
 
 
ansformPepline tp = new TransformPepline(); 
  //添加輸入輸出 
.addSource(fileSourceFunction) 
              .addSink(printSinkFunction) 
              .addSink(listSinkFunction); 
  //定制轉換動作 
      tp.addSourceMapFunction(new Function<String, String>() { 
          SqlTransExecManager sqlTransExecManager = new SqlTransExecManager().build(); 
          @Override 
          public String apply(String s) { 
              return sqlTransExecManager.doParseAndTransform(s).getTargetSql(); 
          } 
      }); 
      //執行 
      tp.execute(); 
      System.out.println(listSinkFunction.getLists()); 
  } 

本文轉載自微信公眾號「麒思妙想」，可以通過以下二維碼關注。轉載本文請聯系麒思妙想公眾號。