1 什麼是Apache Beam

Apache Beam是一個開源的統一的大資料程式設計模型，它本身並不提供執行引擎，而是支援各種平臺如GCP Dataflow、Spark、Flink等。通過Apache Beam來定義批處理或流處理，就可以放在各種執行引擎上運行了。

目前支援的SDK語言也很豐富，有Java、Python、Go等。

1.1 一些基礎概念

• PCollection：可理解為資料包，資料處理就是在對各種PCollection進行轉換和處理。

• PTransform：代表資料處理，用來定義資料是怎麼被處理的，用來處理PCollection。

• Pipeline：流水線，是由PTransform和PCollection組成的集合，可以理解為它定義了資料處理從源到目標的整個過程。

• Runner：資料處理引擎。

一個最簡單的Pipeline例子如下：

從資料庫讀資料為PCollection，經過轉化成為另一個PCollection，然後寫回到資料庫中去。

可以有多個PTransform處理同一個PCollection：

一個PTransform也可以生成多個PCollection：

2 Java開發初體驗

我們通過使用Java SDK來開發一個WordCount感受一下。

先引入必要的依賴，版本為2.32.0：

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-core</artifactId>
  <version>${beam.version}</version>
</dependency>
<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-direct-java</artifactId>
  <version>${beam.version}</version>
</dependency>
 

寫Java主程式如下：

public class WordCountDirect {
    public static void main(String[] args) {
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);
        PCollection<String> lines = pipeline.apply("read from file",
                TextIO.read().from("pkslow.txt"));

        PCollection<List<String>> wordList = lines.apply(MapElements.via(new SimpleFunction<String, List<String>>() {
            @Override
            public List<String> apply(String input) {
                List<String> result = new ArrayList<>();
               char[] chars = input.toCharArray();

                for (char c:chars) {
                    result.add(String.valueOf(c));
                }

              return result;
            }
        }));

        PCollection<String> words = wordList.apply(Flatten.iterables());

        PCollection<KV<String, Long>> wordCount = words.apply(Count.perElement());

        wordCount.apply(MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
            @Override
            public String apply(KV<String, Long> count) {
                return String.format("%s : %s", count.getKey(), count.getValue());
            }
        })).apply(TextIO.write().to("word-count-result"));

        pipeline.run().waitUntilFinish();
    }
}
 

直接執行，預設是通過DirectRunner來執行的，即在本地即可執行，不用搭建。非常方便開發和測試Pipeline。

整個程式大概流程是：

從pkslow.txt檔案裡讀取所有行，然後將每一行拆分為多個字元，計算每個字元出現的次數，輸出到檔案中word-count-result。

pkslow.txt檔案內容如下：

執行後的結果檔案如下所示：

3 總結

簡單體驗了一下，基於Beam的模型開發還是很簡單，很好理解的。但它在各種平臺上的執行效率如何，就還需要深挖了。

程式碼請檢視：https://github.com/LarryDpk/pkslow-samples