1.概述

在大數據的浪潮之下，技能的更新迭代非常頻頻。受技能開源的影響，大數據開發者提供了非常豐厚的東西。但也由于如此，增加了開發者挑選合適東西的難度。在大數據處理一些疑問的時分，通常運用的技能是多元化的。這徹底取決于事務需求，比方進行批處理的MapReduce，實時流處理的Flink，以及SQL交互的Spark SQL等等。而把這些開源結構，東西，類庫，渠道結合到一同，所需求的工作量以及復雜度，可想而知。這也是大數據開發者對比頭疼的疑問。而今天要共享的即是結合這些資本的一個解決方案，它即是 Apache Beam。

2.內容

Apache Beam 最初叫 Apache Dataflow，由谷歌和其合作伙伴向Apache捐贈了大量的核心代碼，并創立孵化了該項目。該項目的大部分大碼來自于 Cloud Dataflow SDK，其特點有以下幾點：

• 統一數據批處理（Batch）和流處理（Stream）編程的范式
• 能運行在任何可執行的引擎之上

那 Apache Beam到底能解決哪些問題，它的應用場景是什么，下面我們可以通過一張圖來說明，如下圖所示：

通過改圖，我們可以很清晰的看到整個技術的發展流向；一部分是谷歌派系，另一部分則是Apache派系。在開發大數據應用時，我們有時候使用谷歌的框架，API，類庫，平臺等，而有時候我們則使用Apache的，比如：HBase，Flink，Spark等。而我們要整合這些資源則是一個比較頭疼的問題，Apache Beam 的問世，整合這些資源提供了很方便的解決方案。

2.1 Vision

下面，我們通過一張流程圖來看Beam的運行流程，如下圖所示：

通過上圖，我們可以清楚的知道，執行一個流程分以下步驟：

1. End Users：選擇一種你熟悉的編程語言提交應用
2. SDK Writers：該編程語言必須是 Beam 模型支持的
3. Library Writers：轉換成Beam模型的格式
4. Runner Writers：在分布式環境下處理并支持Beam的數據處理管道
5. IO Providers：在Beam的數據處理管道上運行所有的應用
6. DSL Writers：創建一個高階的數據處理管道

2.2 SDK

Beam SDK 提供了一個統一的編程模型，來處理任意規模的數據集，其中包括有限的數據集，無限的流數據。Apache Beam SDK 使用相同的類來表達有限和無限的數據，同樣使用相同的轉換方法對數據進行操作。Beam 提供了多種 SDK，你可以選擇一種你熟悉的來建立數據處理管道，如上述的 2.1 中的圖，我們可以知道，目前 Beam 支持 Java，Python 以及其他待開發的語言。

2.3 Pipeline Runners

在 Beam 管道上運行引擎會根據你選擇的分布式處理引擎，其中兼容的 API 轉換你的 Beam 程序應用，讓你的 Beam 應用程序可以有效的運行在指定的分布式處理引擎上。因而，當運行 Beam 程序的時候，你可以按照自己的需求選擇一種分布式處理引擎。當前 Beam 支持的管道運行引擎有以下幾種：

• Apache Apex
• Apache Flink
• Apache Spark
• Google Cloud Dataflow

3.示例

本示例通過使用 Java SDK 來完成，你可以嘗試運行在不同的執行引擎上。

3.1 開發環境

• 下載安裝 JDK 7 或更新的版本，檢測 JAVA_HOME環境變量
• 下載 Maven 打包環境。

關于上述的安裝步驟，并不是本篇博客的重點，這里筆者就不多贅述了，不明白的可以到官網翻閱文檔進行安裝。

3.2 下載示例代碼

Apache Beam 的源代碼在 Github 有托管，可以到 Github 下載對應的源碼，下載地址：https://github.com/apache/beam

然后，將其中的示例代碼進行打包，命令如下所示：

$ mvn archetype:generate \
      -DarchetypeRepository=https://repository.apache.org/content/groups/snapshots \
      -DarchetypeGroupId=org.apache.beam \
      -DarchetypeArtifactId=beam-sdks-java-maven-archetypes-examples \
      -DarchetypeVersion=LATEST \
      -DgroupId=org.example \
      -DartifactId=word-count-beam \
      -Dversion="0.1" \
      -Dpackage=org.apache.beam.examples \
      -DinteractiveMode=false

此時，命令會創建一個文件夾 word-count-beam，里面包含一個 pom.xml 和相關的代碼文件。命令如下所示：

$ cd word-count-beam/

$ ls
pom.xml    src

$ ls src/main/java/org/apache/beam/examples/
DebuggingWordCount.java    WindowedWordCount.java    common
MinimalWordCount.java    WordCount.java

3.3 運行 WordCount 示例代碼

一個 Beam 程序可以運行在多個 Beam 的可執行引擎上，包括 ApexRunner，FlinkRunner，SparkRunner 或者 DataflowRunner。 另外還有 DirectRunner。不需要特殊的配置就可以在本地執行，方便測試使用。

下面，你可以按需選擇你想執行程序的引擎：

1. 對引擎進行相關配置
2. 使用不同的命令：通過 –runner=<runner>參數指明引擎類型，默認是 DirectRunner；添加引擎相關的參數；指定輸出文件和輸出目錄，當然這里需要保證文件目錄是執行引擎可以訪問到的，比如本地文件目錄是不能被外部集群訪問的。
3. 運行示例程序

3.3.1 Direct

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--inputFile=pom.xml --output=counts" -Pdirect-runner

3.3.2 Apex

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--inputFile=pom.xml --output=counts --runner=ApexRunner" -Papex-runner

3.3.3 Flink-Local

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--runner=FlinkRunner --inputFile=pom.xml --output=counts" -Pflink-runner

3.3.4 Flink-Cluster

$ mvn package exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--runner=FlinkRunner --flinkMaster=<flink master> --filesToStage=target/word-count-beam-bundled-0.1.jar \
                  --inputFile=/path/to/quickstart/pom.xml --output=/tmp/counts" -Pflink-runner

然后，你可以通過訪問?http://<flink master>:8081 來監測運行的應用程序。

3.3.5 Spark

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--runner=SparkRunner --inputFile=pom.xml --output=counts" -Pspark-runner

3.3.6 Dataflow

$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
     -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://<your-gcs-bucket>/tmp \
                  --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://<your-gcs-bucket>/counts" \
     -Pdataflow-runner

3.4 運行結果

當程序運行完成后，你可以看到有多個文件以 count 開頭，個數取決于執行引擎的類型。當你查看文件的內容的時候，每個唯一的單詞后面會顯示其出現次數，但是前后順序是不固定的，也是分布式引擎為了提高效率的一種常用方式。

3.4.1 Direct

$ ls counts*

$ more counts*
api: 9
bundled: 1
old: 4
Apache: 2
The: 1
limitations: 1
Foundation: 1
...

3.4.2 Apex

$ cat counts*
BEAM: 1
have: 1
simple: 1
skip: 4
PAssert: 1
...

3.4.3 Flink-Local

$ ls counts*

$ more counts*
The: 1
api: 9
old: 4
Apache: 2
limitations: 1
bundled: 1
Foundation: 1
...

3.4.4 Flink-Cluster

$ ls /tmp/counts*

$ more /tmp/counts*
The: 1
api: 9
old: 4
Apache: 2
limitations: 1
bundled: 1
Foundation: 1
...

3.4.5 Spark

$ ls counts*

$ more counts*
beam: 27
SF: 1
fat: 1
job: 1
limitations: 1
require: 1
of: 11
profile: 10
...

3.4.6 Dataflow

$ gsutil ls gs://<your-gcs-bucket>/counts*

$ gsutil cat gs://<your-gcs-bucket>/counts*
feature: 15
smother'st: 1
revelry: 1
bashfulness: 1
Bashful: 1
Below: 2
deserves: 32
barrenly: 1
...

4.總結

Apache Beam 首要對于抱負并行的數據處理使命，并通過把數據集拆分多個子數據集，讓每個子數據集可以被單獨處理，然后完成全體數據集的并行化處理。當然，也可以用 Beam 來處理抽取，變換和加載使命和數據集成使命（一個ETL進程）。進一步將數據從不一樣的存儲介質中或許數據源中讀取，變換數據格式，最后加載到新的系統中。