1. 概述
在本教程中摩渺，我們將介紹 Apache Beam 并探討其基本概念。
我們將首先演示使用 Apache Beam 的用例和好處剂邮，然后介紹基本概念和術(shù)語摇幻。之后，我們將通過一個簡單的例子來說明 Apache Beam 的所有重要方面挥萌。
2. Apache Beam是個啥?
Apache Beam（Batch+strEAM）是一個用于批處理和流式數(shù)據(jù)處理作業(yè)的統(tǒng)一編程模型绰姻。它提供了一個軟件開發(fā)工具包，用于定義和構(gòu)建數(shù)據(jù)處理管道以及執(zhí)行這些管道的運行程序引瀑。
Apache Beam旨在提供一個可移植的編程層狂芋。事實上，Beam管道運行程序?qū)?shù)據(jù)處理管道轉(zhuǎn)換為與用戶選擇的后端兼容的API憨栽。目前帜矾，支持這些分布式處理后端有：
- Apache Apex
- Apache Flink
- Apache Gearpump (incubating)
- Apache Samza
- Apache Spark
- Google Cloud Dataflow
- Hazelcast Jet
3. 為啥選擇 Apache Beam
Apache Beam 將批處理和流式數(shù)據(jù)處理融合在一起，而其他組件通常通過單獨的 API 來實現(xiàn)這一點屑柔。因此屡萤，很容易將流式處理更改為批處理，反之亦然掸宛，例如死陆，隨著需求的變化。
Apache Beam 提高了可移植性和靈活性唧瘾。我們關(guān)注的是邏輯措译，而不是底層的細節(jié)迫像。此外，我們可以隨時更改數(shù)據(jù)處理后端瞳遍。
Apache Beam 可以使用 Java闻妓、Python、Go和 Scala等SDK掠械。事實上由缆，團隊中的每個人都可以使用他們選擇的語言。
4. 基本概念
使用 Apache Beam猾蒂，我們可以構(gòu)建工作流圖（管道）并執(zhí)行它們均唉。編程模型中的關(guān)鍵概念是：
PCollection–表示可以是固定批處理或數(shù)據(jù)流的數(shù)據(jù)集
PTransform–一種數(shù)據(jù)處理操作，它接受一個或多個 PCollections 并輸出零個或多個 PCollections肚菠。
Pipeline–表示 PCollection 和 PTransform 的有向無環(huán)圖舔箭，因此封裝了整個數(shù)據(jù)處理作業(yè)。
PipelineRunner–在指定的分布式處理后端上執(zhí)行管道蚊逢。
簡單地說层扶，PipelineRunner 執(zhí)行一個管道，管道由 PCollection 和 PTransform 組成烙荷。
5. 字數(shù)統(tǒng)計示例
現(xiàn)在我們已經(jīng)學習了 Apache Beam 的基本概念镜会，讓我們設(shè)計并測試一個單詞計數(shù)任務(wù)。
5.1 建造梁式管道
設(shè)計工作流圖是每個 Apache Beam 作業(yè)的第一步终抽，單詞計數(shù)任務(wù)的步驟定義如下：
1.從原文中讀課文戳表。
2.把課文分成單詞表。
3.所有單詞都小寫昼伴。
4.刪去標點符號匾旭。
5.過濾停止語。
6.統(tǒng)計唯一單詞數(shù)量圃郊。
為了實現(xiàn)這一點价涝，我們需要使用 PCollection 和 PTransform 抽象將上述步驟轉(zhuǎn)換為 管道 。
5.2. 依賴
在實現(xiàn)工作流圖之前描沟，先添加 Apache Beam的依賴項 到我們的項目：
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>${beam.version}</version>
</dependency>
Beam管道運行程序依賴于分布式處理后端來執(zhí)行任務(wù)飒泻。我們添加 DirectRunner 作為運行時依賴項：
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>${beam.version}</version>
<scope>runtime</scope>
</dependency>
與其他管道運行程序不同鞭光，DirectRunner 不需要任何額外的設(shè)置吏廉，這對初學者來說是個不錯的選擇。
5.3. 實現(xiàn)
Apache Beam 使用 Map-Reduce 編程范式 ( 類似 Java Stream)惰许。講下面內(nèi)容之前席覆，最好 對 reduce(), filter(), count(), map(), 和 flatMap() 有個基礎(chǔ)概念和認識。
首先要做的事情就是 創(chuàng)建管道：
PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
六步單詞計數(shù)任務(wù)：
PCollection<KV<String, Long>> wordCount = p
.apply("(1) Read all lines",
TextIO.read().from(inputFilePath))
.apply("(2) Flatmap to a list of words",
FlatMapElements.into(TypeDescriptors.strings())
.via(line -> Arrays.asList(line.split("\\s"))))
.apply("(3) Lowercase all",
MapElements.into(TypeDescriptors.strings())
.via(word -> word.toLowerCase()))
.apply("(4) Trim punctuations",
MapElements.into(TypeDescriptors.strings())
.via(word -> trim(word)))
.apply("(5) Filter stopwords",
Filter.by(word -> !isStopWord(word)))
.apply("(6) Count words",
Count.perElement());
apply() 的第一個（可選）參數(shù)是一個String汹买，它只是為了提高代碼的可讀性佩伤。下面是上述代碼中每個 apply() 的作用：
- 首先聊倔，我們使用 TextIO 逐行讀取輸入文本文件。
- 將每一行按空格分開生巡，把它映射到一個單詞表上耙蔑。
- 單詞計數(shù)不區(qū)分大小寫，所以我們將所有單詞都小寫孤荣。
- 之前甸陌，我們用空格分隔行，但是像“word盐股！“和”word钱豁？"這樣的，就需要刪除標點符號疯汁。
- 像“is”和“by”這樣的停止詞在幾乎每一篇英語文章中都很常見牲尺，所以我們將它們刪除。
- 最后幌蚊，我們使用內(nèi)置函數(shù) Count.perElement() 計算唯一單詞數(shù)量谤碳。
如前所述，管道是在分布式后端處理的溢豆。不可能在內(nèi)存中的PCollection上迭代估蹄，因為它分布在多個后端。相反沫换，我們將結(jié)果寫入外部數(shù)據(jù)庫或文件臭蚁。
首先，我們將PCollection轉(zhuǎn)換為String讯赏。然后垮兑，使用TextIO編寫輸出：
wordCount.apply(MapElements.into(TypeDescriptors.strings())
.via(count -> count.getKey() + " --> " + count.getValue()))
.apply(TextIO.write().to(outputFilePath));
現(xiàn)在管道 已經(jīng)定義好了，接下來做個簡單的測試漱挎。
5.4. 運行測試
到目前為止系枪，我們已為單詞計數(shù)任務(wù)定義了管道，現(xiàn)在運行管道：
p.run().waitUntilFinish();
在這行代碼中磕谅，Apache Beam 將把我們的任務(wù)發(fā)送到多個 DirectRunner 實例私爷。因此，最后將生成幾個輸出文件膊夹。它們將包含以下內(nèi)容：
...
apache --> 3
beam --> 5
rocks --> 2
...
在 Apache Beam 中定義和運行分布式作業(yè)是如此地簡單衬浑。為了進行比較，單詞計數(shù)實現(xiàn)在 Apache Spark, Apache Flink 和 Hazelcast-Jet 上也有
6. 結(jié)語
在本教程中放刨，我們了解了 Apache Beam 是什么工秩，以及它為什么比其他選擇更受歡迎。我們還通過一個單詞計數(shù)示例演示了 Apache Beam 的基本概念。
如果你覺得文章還不錯助币，記得關(guān)注公眾號： 鍋外的大佬
鍋外的大佬博客
