作者：張海濤

關(guān)于 Apache Beam 實(shí)戰(zhàn)指南系列文章

隨著大數(shù)據(jù) 2.0 時代悄然到來，大數(shù)據(jù)從簡單的批處理擴(kuò)展到了實(shí)時處理、流處理、交互式查詢和機(jī)器學(xué)習(xí)應(yīng)用。近年來涌現(xiàn)出諸多大數(shù)據(jù)應(yīng)用組件，如 HBase、Hive、Kafka、Spark、Flink 等。開發(fā)者經(jīng)常要用到不同的技術(shù)、框架、API、開發(fā)語言和 SDK 來應(yīng)對復(fù)雜應(yīng)用的開發(fā)，這大大增加了選擇合適工具和框架的難度，開發(fā)者想要將所有的大數(shù)據(jù)組件熟練運(yùn)用幾乎是一項(xiàng)不可能完成的任務(wù)。

面對這種情況，Google 在 2016 年 2 月宣布將大數(shù)據(jù)流水線產(chǎn)品(Google DataFlow)貢獻(xiàn)給 Apache 基金會孵化，2017 年 1 月 Apache 對外宣布開源 Apache Beam，2017 年 5 月迎來了它的第一個穩(wěn)定版本 2.0.0。在國內(nèi)，大部分開發(fā)者對于 Beam 還缺乏了解，社區(qū)中文資料也比較少。

一.概述

其他行業(yè)問咱們 IT 具體干什么的，很多 IT 人員會自嘲自己就是“搬磚”(此處將復(fù)制代碼稱為搬磚)的民工。過了兩天 GitHub 出現(xiàn)自動寫代碼的人工智能，IT 程序員深深嘆了一口氣說道“完了要失業(yè)了，代碼沒得搬了”。其實(shí)從入行 IT 那一刻起，不管我們做前端、服務(wù)端、底層架構(gòu)等任何崗位，其實(shí)我們都是為數(shù)據(jù)服務(wù)的服務(wù)人員(注：不是說從民工轉(zhuǎn)崗到服務(wù)員了)：把數(shù)據(jù)從后端搬到前端，把前端數(shù)據(jù)再寫入數(shù)據(jù)庫。盡管編程語言從 C、C++、C#、JAVA、Python 不停變化，為了適應(yīng)時代背景框架也是千變?nèi)f化，我們拼命從“亞馬遜熱帶雨林”一直學(xué)到“地中海”。

然后 Apache Beam 這個一統(tǒng)“地中海”的框架出現(xiàn)了。Apache Beam 不光統(tǒng)一了數(shù)據(jù)源，還統(tǒng)一了流批計算。在這個數(shù)據(jù)傳輸過程中有一條核心的技術(shù)就是管道(Pipeline)，不管是 Strom，F(xiàn)link ，Beam 它都是核心。在這條管道中可以對數(shù)據(jù)進(jìn)行過濾、凈化、清洗、合并、分流以及各種實(shí)時計算操作。

本文會詳細(xì)介紹如何設(shè)計 Apache Beam 管道、管道設(shè)計工具介紹、源碼和案例分析，普及和提升大家對 Apache Beam 管道的認(rèn)知。

二.怎樣設(shè)計好自己的管道?

設(shè)計管道注意事項(xiàng)

 

 

圖 2-1 簡單管道

1. 你輸入的數(shù)據(jù)存儲在那里?

首先要確定你要構(gòu)造幾條數(shù)據(jù)源，在 Beam 可以構(gòu)建多條，構(gòu)建之前可以選擇自己的 SDK 的 IO。

2. 你的數(shù)據(jù)類型是什么樣的?

Beam 提供的是鍵值對的數(shù)據(jù)類型，你的數(shù)據(jù)可能是日志文本、格式化設(shè)備事件、數(shù)據(jù)庫的行，所以在 PCollection 就應(yīng)該確定數(shù)據(jù)集的類型。

3. 你想怎么處理數(shù)據(jù)?

對數(shù)據(jù)進(jìn)行轉(zhuǎn)換、過濾處理、窗口計算、SQL 處理等。 在管道中提供了通用的 ParDo 轉(zhuǎn)換類，算子計算以及 BeamSQL 等操作。

4. 你打算把數(shù)據(jù)最后輸出到哪里去?

在管道末尾進(jìn)行 Write 寫入操作，把數(shù)據(jù)最后寫入你自己想存放或最后流向的地方。

管道的幾種玩法

1. 分支管道：多次轉(zhuǎn)換，處理相同的數(shù)據(jù)集

 

 

圖 2-2-1 多次轉(zhuǎn)換處理相同數(shù)據(jù)示意圖

描述：例如上圖 2-1-1 圖所示，從一個數(shù)據(jù)庫的表讀取或轉(zhuǎn)換數(shù)據(jù)集，然后從數(shù)據(jù)集中分別找找以字母“A”開頭的數(shù)據(jù)放入一個分支數(shù)據(jù)集中，如果以字母“B”開頭的數(shù)據(jù)放入另一個分支數(shù)據(jù)集中，最終兩個數(shù)據(jù)集進(jìn)行隔離處理。

數(shù)據(jù)集：

// 為了演示顯示內(nèi)存數(shù)據(jù)集
final List LINES = Arrays.asList(
"Aggressive",
"Bold",
"Apprehensive",
"Brilliant");

示例代碼：

�pPCollection dbRowCollection = ...;// 這個地方可以讀取任何數(shù)據(jù)源。
PCollection aCollection = dbRowCollection.apply("aTrans", ParDo.of(new DoFn(){
@ProcessElement
public void processElement(ProcessContext c) {
if(c.element().startsWith("A")){// 查找以 "A" 開頭的數(shù)據(jù)
c.output(c.element());
System.out.append("A 開頭的單詞有:"+c.element()+"\r");
}
}
}));
PCollection bCollection = dbRowCollection.apply("bTrans", ParDo.of(new DoFn(){
@ProcessElement
public void processElement(ProcessContext c) {
if(c.element().startsWith("B")){// 查找以 "A" 開頭的數(shù)據(jù)
c.output(c.element());
System.out.append("B 開頭的單詞有:"+c.element()+"\r");
}
}
}));

最終結(jié)果展示：

A 開頭的單詞有:Aggressive
B 開頭的單詞有:Bold
A 開頭的單詞有:Apprehensive
B 開頭的單詞有:Brilliant

原示例代碼地址 : pipelineTest2_1

2. 分支管道：一次轉(zhuǎn)換，輸出多個數(shù)據(jù)集

 

 

圖 2-2-2 一次轉(zhuǎn)換多個輸出示意圖

描述：根據(jù)圖 2-2-1 和圖 2-2-2 圖中可以看出，他們以不同的方式執(zhí)行著相同的操作，圖 2-2-1 中的管道包含兩個轉(zhuǎn)換，用于處理同一輸入中的元素 PCollection。一個轉(zhuǎn)換使用以下邏輯：

if（以'A'開頭）{outputToPCollectionA}

另一個轉(zhuǎn)換為

if（以'B'開頭）{outputToPCollectionB}

因?yàn)槊總�轉(zhuǎn)換讀取整個輸入 PCollection，所以輸入中的每個元素都會 PCollection 被處理兩次。

圖 2-2-2 中的管道以不同的方式執(zhí)行相同的操作 - 只有一個轉(zhuǎn)換使用以下邏輯：

if（以'A'開頭）{outputToPCollectionA} else if（以'B'開頭）{outputToPCollectionB}

其中輸入中的每個元素都 PCollection 被處理一次。

數(shù)據(jù)集：同 2-1-1 數(shù)據(jù)集

示例代碼：

�p// 定義兩個 TupleTag，每個輸出一個。
final TupleTag startsWithATag = new TupleTag(){};
final TupleTag startsWithBTag = new TupleTag(){};
PCollectionTuple mixedCollection =
dbRowCollection.apply(ParDo
.of(new DoFn() {
@ProcessElement
public void processElement(ProcessContext c) {
if (c.element().startsWith("A")) {
// 返回首字母帶有 "A" 的數(shù)據(jù)集。
c.output(c.element());
} else if(c.element().startsWith("B")) {
// // 返回首字母帶有 "B" 的數(shù)據(jù)集。
c.output(startsWithBTag, c.element());
}
}
})
// Specify main output. In this example, it is the output
// with tag startsWithATag.
.withOutputTags(startsWithATag,
// Specify the output with tag startsWithBTag, as a TupleTagList.
TupleTagList.of(startsWithBTag)));
// Get subset of the output with tag startsWithATag.
mixedCollection.get(startsWithATag).apply(...);
// Get subset of the output with tag startsWithBTag.
mixedCollection.get(startsWithBTag).apply(...);

如果每個元素的轉(zhuǎn)換計算非常耗時，則使用其他輸出會更有意義，因?yàn)橐淮涡赃^濾全部數(shù)據(jù)，比全部數(shù)據(jù)過濾兩次從性能上和轉(zhuǎn)換上都存在一定程度上提升，數(shù)據(jù)量越大越明顯。

最終結(jié)果展示：

�p復(fù)制代碼A 開頭的單詞有:ApprehensiveA 開頭的單詞有:AggressiveB 開頭的單詞有:BrilliantB 開頭的單詞有:Bold

原示例代碼地址 : pipelineTest2_2

3. 合并管道：多個數(shù)據(jù)集，合并成一個管道輸出

 

 

圖 2-2-3 多數(shù)據(jù)集合并輸出圖

描述：

上圖 2-2-3 是接圖 2-2-1 的繼續(xù)，把帶“A” 的數(shù)據(jù)和帶“B” 字母開頭的數(shù)據(jù)進(jìn)行合并到一個管道。

這個地方注意點(diǎn)是 Flatten 用法必須兩個數(shù)據(jù)的數(shù)據(jù)類型相同。

數(shù)據(jù)集：

// 為了演示顯示內(nèi)存數(shù)據(jù)集
final List LINESa = Arrays.asList(
"Aggressive",
"Apprehensive");
final List LINESb = Arrays.asList(
"Bold",
"Brilliant");

示例代碼：

// 將兩個 PCollections 與 Flatten 合并
PCollectionList collectionList = PCollectionList.of(aCollection).and(bCollection);
PCollection mergedCollectionWithFlatten = collectionList
.apply(Flatten.pCollections());
// 繼續(xù)合并新的 PCollection
mergedCollectionWithFlatten.apply(...);

結(jié)果展示：

合并單詞單詞有:
Aggressive
Brilliant
Apprehensive
Bold

原示例代碼地址 : pipelineTest2_3

4. 合并管道：多個數(shù)據(jù)源，鏈接合并一個管道輸出

 

 

圖 2-2-4 多數(shù)據(jù)源合并輸出圖

描述：

你的管道可以從一個或多個源讀取或輸入。如果你的管道從多個源讀取并且這些源中的數(shù)據(jù)相關(guān)聯(lián)，則將輸入連接在一起會很有用。在上面的圖 2-2-4 所示的示例中，管道從數(shù)據(jù)庫表中讀取名稱和地址，并從 Kafka 主題中讀取名稱和訂單號。然后管道 CoGroupByKey 用于連接此信息，其中鍵是名稱 ; 結(jié)果 PCollection 包含名稱，地址和訂單的所有組合。

示例代碼：

PCollection> userAddress = pipeline.apply(JdbcIO.>read()...);
PCollection> userOrder = pipeline.apply(KafkaIO.read()...);
final TupleTag addressTag = new TupleTag();
final TupleTag orderTag = new TupleTag();
// 將集合值合并到 CoGbkResult 集合中。
PCollection> joinedCollection =
KeyedPCollectionTuple.of(addressTag, userAddress)
.and(orderTag, userOrder)
.apply(CoGroupByKey.create());
joinedCollection.apply(...);

管道的設(shè)計工具

對于管道的設(shè)計不光用代碼去實(shí)現(xiàn)，也可以用視圖工具�，F(xiàn)在存在的有兩種一種是拓藍(lán)公司出品叫 Talend Big Data Studio，另一種就是免費(fèi)開源的視圖設(shè)計工具 kettle-beam 。

 

 

三.怎樣創(chuàng)建你的管道

Apache Beam 程序從頭到尾就是處理數(shù)據(jù)的管道。本小節(jié)使用 Apache Beam SDK 中的類構(gòu)建管道，一個完整的 Apache Beam 管道構(gòu)建流程如下：

首先創(chuàng)建一個 Pipeline 對象。

不管是數(shù)據(jù)做任何操作，如“ 讀取”或“ 創(chuàng)建”及轉(zhuǎn)換都要為管道創(chuàng)建 PCollection 一個或多個的數(shù) 據(jù)集(PCollection****)。

在 Apache Beam 的管道中你可以對數(shù)據(jù)集 PCollection 做任何操作，例如轉(zhuǎn)換數(shù)據(jù)格式，過濾，分組，分析或以其他方式處理數(shù)據(jù)中的每一個元素。每個轉(zhuǎn)換都會創(chuàng)建一個新輸出數(shù)據(jù)集 PCollection，當(dāng)然你可以在處理完成之前進(jìn)行做任何的轉(zhuǎn)換處理。

把你認(rèn)為最終處理完成的數(shù)據(jù)集寫或以其他方式輸出最終的存儲地方。

最后運(yùn)行管道。

創(chuàng)建管道對象

每一個 Apache Beam 程序都會從創(chuàng)建管道(Pipeline)對象開始。

在 Apache Beam SDK，每一個管道都是一個獨(dú)立的實(shí)體，管道的數(shù)據(jù)集也都封裝著它的數(shù)據(jù)和對應(yīng)的數(shù)據(jù)類型(在 Apache Beam 中有對應(yīng)的數(shù)據(jù)轉(zhuǎn)換類型包)。最后把數(shù)據(jù)進(jìn)行用于各種轉(zhuǎn)換操作。

在創(chuàng)建的管道的時候需要設(shè)置管道選項(xiàng) PipelineOptions，有兩種創(chuàng)建方式第一種是無參數(shù)和一種有參數(shù)的。具體兩種有什么不同呢? 無參數(shù)的可以在程序中指定相應(yīng)的管道選項(xiàng)參數(shù)，如顯示設(shè)置執(zhí)行大數(shù)據(jù)引擎參數(shù)。有參數(shù)的就可以在提交 Apache Beam jar 程序的時候進(jìn)行用 Shell 腳本的方式后期設(shè)置管道對應(yīng)的參數(shù)。

具體示例如下：

無參數(shù)

// 首先定義管道的選項(xiàng)
PipelineOptions options = PipelineOptionsFactory.create();
options.setRunner(DirectRunner.class); // 顯示設(shè)置執(zhí)行大數(shù)據(jù)引擎
// 創(chuàng)建管道實(shí)體對象
Pipeline p = Pipeline.create(options);

有參數(shù)

PipelineOptions options =
PipelineOptionsFactory.fromArgs(args).withValidation().create();

提交設(shè)置參數(shù)的格式如下：

--

=

將數(shù)據(jù)讀入你的管道

創(chuàng)建 PCollection 的初始值，請讀取外部的數(shù)據(jù)源及指定的本地數(shù)據(jù)。例如讀取數(shù)據(jù)庫，文本文件，流數(shù)據(jù)等等，現(xiàn)在 Apache Beam java SDKS 支持 33 種數(shù)據(jù)源，正在接入集成的有 7 種，Python 13 種，正在集成的 1 種。基本覆蓋了 IT 行業(yè)的一切數(shù)據(jù)源。例如讀取文本數(shù)據(jù)我們可以用 TextIO.Read 的方法進(jìn)行讀取數(shù)據(jù)。轉(zhuǎn)換應(yīng)用于管道對象 p 中。并且返回對應(yīng)格式的數(shù)據(jù)集 PCollection：

PCollection lines = p.apply(
"ReadLines", TextIO.read().from("/home/inputData.txt"));

注意：在 Apache Beam 程序執(zhí)行中，Beam 程序 2.2.0 以前版本不支持 Windows 如：D\inputData.txt 路徑格式。只支持 linux 路徑格式，及其他如 HDFS 等存儲路徑。

已經(jīng)支持的數(shù)據(jù)源統(tǒng)計如下表：

 

 

將管道數(shù)據(jù)轉(zhuǎn)換為處理的格式

很多時候直接從數(shù)據(jù)源讀取的數(shù)據(jù)不會直接流入目標(biāo)存儲。大部分需要進(jìn)行數(shù)據(jù)格式的轉(zhuǎn)換，數(shù)據(jù)的清洗，數(shù)據(jù)的過濾，數(shù)據(jù)的關(guān)聯(lián)處理，數(shù)據(jù)累加操作等。這里需要對源數(shù)據(jù)進(jìn)行處理，處理完成的數(shù)據(jù)處理流入目標(biāo)存儲外還可以進(jìn)行當(dāng)作參數(shù)一樣，傳遞并繼續(xù)應(yīng)用到管道中。

以下示例代碼為 把一串?dāng)?shù)據(jù)通過轉(zhuǎn)換操作賦值給 words ， 然后再把 words 再次傳遞到下一個操作應(yīng)用，再進(jìn)一步進(jìn)行操作的處理工作。

PCollection lines = p.apply(
"ReadLines", TextIO.read().from("/home/inputData.txt"));

編寫或輸出管道最終輸出數(shù)據(jù)的地方

經(jīng)過一些列的清洗、過濾、關(guān)聯(lián)、轉(zhuǎn)換處理工作后的數(shù)據(jù)，最終都會通過 SDKIO 進(jìn)行寫入管道外的存儲或著數(shù)據(jù)庫表。然而這種寫入操作大部分都是在管道的末尾端進(jìn)行操作的。

如下面代碼示例，就是把管道的數(shù)據(jù)通過 Apache Beam 中的 TextIO.Write 寫入 Linux 的文本文件 test.txt 中。

PCollection filteredWords = ...;
filteredWords.apply("WriteMyFile", TextIO.write().to("/home/test.txt"));

運(yùn)行你的管道

構(gòu)建管道后，使用 run 方法最后執(zhí)行管道。 管道以異步方式執(zhí)行的。 寫完這一句代碼后你就可以把自己的程序用 Jenkins 進(jìn)行編譯并提交給運(yùn)行管道平臺，最終有管道執(zhí)行平臺來運(yùn)行。

運(yùn)行代碼示例：

p.run();

處理異步執(zhí)行的方式，還有同步執(zhí)行方式，是在 run 方法后面加個看守方法 waitUntilFinish。具體代碼如下：

p.run().waitUntilFinish();

四.怎樣測試你的管道

Apache Beam 管道開發(fā)中最后的測試在整個開發(fā)中也是非常重要的一個環(huán)節(jié)。Apache Beam 的代碼程序不必要每次都進(jìn)行遠(yuǎn)程構(gòu)建執(zhí)行到 Flink 集群上，因?yàn)楣艿来�碼的錯誤及 Bug 的修改在本地能更好的調(diào)試，然而每次構(gòu)建到遠(yuǎn)程上面去執(zhí)行是非常麻煩的事情。Apache Beam 提供 DirectRunner ，一個用于本地測試的執(zhí)行引擎。

使用 DirectRunner 測試管道的時候，你可以用小規(guī)模的數(shù)據(jù)進(jìn)行測試。此外你如果開發(fā)機(jī)器上裝了本地的 flink ，也可以指定本地的 Flink 執(zhí)行。 例如測試一個簡單的轉(zhuǎn)換函數(shù) DoFn，符合變換，數(shù)據(jù)源輸入到管道尾端數(shù)據(jù)輸出等操作。

注意點(diǎn)：DirectRunner 是用于管道或 Apache Beam 程序 本地開發(fā)調(diào)試測試的 數(shù)據(jù)執(zhí)行引擎，不可以用于真正生產(chǎn)環(huán)境中運(yùn)行。否則程序執(zhí)行性能會大大降低，這里有坑要避開。

測試單個 Pipeline 步驟

我們開發(fā)完成管道 Beam 程序后需要本地測試，Beam SDK for Java 提供了一種方便的方法來測試 TestPipeline 的封裝類。 在 Beam SDK testing 包中。

它的使用操作方法：

1. 創(chuàng)建一個 TestPipeline。

2. 創(chuàng)建一些已知的靜態(tài)測試數(shù)據(jù)，也稱為內(nèi)存數(shù)據(jù)，真正應(yīng)用基本是流或批數(shù)據(jù)。

3. 使用 Create 方法創(chuàng)建 PCollection 輸入數(shù)據(jù)。

4. 使用 Apply 方法進(jìn)行數(shù)據(jù)的轉(zhuǎn)換處理并且返回指定的 PCollection。

5. 最后使用 PAssert 去驗(yàn)證輸出的結(jié)果是否為預(yù)期結(jié)果值。

測試實(shí)戰(zhàn)示例

Apache Beam 中簡單的管道單元測試實(shí)例。

�ppublic class CountTest {
// 創(chuàng)建靜態(tài)的內(nèi)存數(shù)據(jù)
static final String[] WORDS_ARRAY = new String[] {
"hi", "there", "hi", "hi", "sue", "bob",
"hi", "sue", "", "", "ZOW", "bob", ""};
static final List WORDS = Arrays.asList(WORDS_ARRAY);
public void testCount() {
// 創(chuàng)建一個測試管道.
Pipeline p = TestPipeline.create();
// 創(chuàng)建一個輸入數(shù)據(jù)集.
PCollection input = p.apply(Create.of(WORDS)).setCoder(StringUtf8Coder.of());
// 添加轉(zhuǎn)換統(tǒng)計單詞個數(shù).
PCollection> output =
input.apply(Count.perElement());
// 驗(yàn)證結(jié)果.
PAssert.that(output)
.containsInAnyOrder(
KV.of("hi", 4L),
KV.of("there", 1L),
KV.of("sue", 2L),
KV.of("bob", 2L),
KV.of("", 3L),
KV.of("ZOW", 1L));
// 運(yùn)行整個管道.
p.run();
}

端到端的測試管道

端到端的測試，主要針對輸入端和輸出端兩端的測試。要測試整個管道，請執(zhí)行以下操作：

創(chuàng)建一個 Beam 測試 SDK 中所提供的 TestPipeline 實(shí)例。

對于多步驟數(shù)據(jù)流水線中的每個輸入數(shù)據(jù)源，創(chuàng)建相對應(yīng)的靜態(tài)(Static)測試數(shù)據(jù)集。

使用 Create Transform，將所有的這些靜態(tài)測試數(shù)據(jù)集轉(zhuǎn)換成 PCollection 作為輸入數(shù)據(jù)集。

按照真實(shí)數(shù)據(jù)流水線邏輯，調(diào)用所有的 Transforms 操作。

在數(shù)據(jù)流水線中所有應(yīng)用到 Write Transform 的地方，都使用 PAssert 來替換這個 Write Transform，并且驗(yàn)證輸出的結(jié)果是否我們期望的結(jié)果相匹配

由于端到端測試跟單個 Pipeline 步驟相似就不在舉示例代碼。其實(shí)開發(fā)過程中本地調(diào)試打斷點(diǎn)，寫日志測試也是更快解決問題的一個辦法。

五. Apache Beam 的管道源碼解析

Apache Beam Pipeline 源碼解析

管道源代碼主類是比較簡單的，本文針對 Pipeline.java 進(jìn)行解析。

1. 定義管道參數(shù)及管道創(chuàng)建

在管道創(chuàng)建首先可以定義管道的選項(xiàng)，例如 Beam 作業(yè)程序的名稱、唯一標(biāo)識、運(yùn)行引擎平臺等，當(dāng)然也可以提交引擎平臺用命令指定也可以。然后實(shí)例化一個管道對象。

源碼示例如下：

�pPipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);

2. 讀取數(shù)據(jù)源

讀取要處理的數(shù)據(jù)，有文本數(shù)據(jù)，結(jié)構(gòu)化數(shù)據(jù)和非結(jié)構(gòu)化數(shù)據(jù)以及流數(shù)據(jù)。作為數(shù)據(jù)處理的源數(shù)據(jù)。

源碼示例如下：

PCollection lines =
p.apply(TextIO.read().from("gs://bucket/dir/file*.txt"));

3. 進(jìn)行數(shù)據(jù)處理操作

在管道里面可以進(jìn)行窗口操作、函數(shù)操作、原子操作以及 SQL 操作。

數(shù)據(jù)統(tǒng)計的源碼示例：

PCollection> wordCounts =allLines
.apply(ParDo.of(new ExtractWords()))
.apply(new Count());

4. 輸出結(jié)果及運(yùn)行

源代碼示例：

PCollection formattedWordCounts =
wordCounts.apply(ParDo.of(new FormatCounts()));
formattedWordCounts.apply(TextIO.write().to("gs://bucket/dir/counts.txt"));
p.run();

六.管道實(shí)戰(zhàn)案例

案例場景描述

隨著人工智能 的不斷發(fā)展，AI Cloud 在銀行加快落地，安防 AI 碎片化的應(yīng)用場景遍地開花。本文結(jié)合銀行營業(yè)網(wǎng)點(diǎn)的業(yè)務(wù)，介紹管道案例實(shí)戰(zhàn)。

以銀行的員工脫離崗檢測中的行為分析數(shù)據(jù)預(yù)處理為例。我們?nèi)ャy行辦理業(yè)務(wù)過程中，首先要取號，然后叫號。叫號提示會對接系統(tǒng)形成一條消息回傳后臺，但是有時候正常辦理業(yè)務(wù)期間有柜臺營業(yè)員出去，然后很久才回來。這個時候攝像頭會根據(jù)柜臺離崗時間自動 AI 行為分析生成報警處理。

案例業(yè)務(wù)架構(gòu)流程

 

 

叫號報警和行為分析報警產(chǎn)生的數(shù)據(jù)通過營業(yè)網(wǎng)點(diǎn)進(jìn)行上報。

上傳網(wǎng)關(guān)集群，網(wǎng)關(guān)集群進(jìn)行轉(zhuǎn)換消息格式壓縮消息。

消息流入消息中心等待消費(fèi)，消息中心再次起著消峰作用。

用 Beam 管道的時間窗口特性、流合并處理特性進(jìn)行消息消費(fèi)處理

消息進(jìn)入大數(shù)據(jù)實(shí)時分析處理平臺處理應(yīng)用消息。

案例示例核心代碼

1. 本案例為了節(jié)約閱讀時間，采用靜態(tài)數(shù)據(jù)

�p // 創(chuàng)建管道工廠
PipelineOptions options = PipelineOptionsFactory.create();
// 顯式指定 PipelineRunner：FlinkRunner 必須指定如果不制定則為本地
options.setRunner(DirectRunner.class); // 生產(chǎn)環(huán)境關(guān)閉
// options.setRunner(FlinkRunner.class); // 生成環(huán)境打開
Pipeline pipeline = Pipeline.create(options);// 設(shè)置相關(guān)管道
// 為了演示顯示內(nèi)存數(shù)據(jù)集
// 叫號數(shù)據(jù)
final List> txtnoticelist = Arrays.asList(KV.of("DS-2CD2326FWDA3-I", "101 號顧客請到 3 號柜臺 "), KV.of("DS-2CD2T26FDWDA3-IS", "102 號顧客請到 1 號柜臺 "),
KV.of("DS-2CD6984F-IHS", "103 號顧客請到 4 號柜臺 "),
KV.of("DS-2CD7627HWD-LZS", "104 號顧客請到 2 號柜臺 "));
//AI 行為分析消息
final List> aimessagelist = Arrays.asList(KV.of("DS-2CD2326FWDA3-I",
"CMOS 智能半球網(wǎng)絡(luò)攝像機(jī), 山東省濟(jì)南市解放路支行 3 號柜,type=2,display_image=no"),
KV.of("DS-2CD2T26FDWDA3-IS", "CMOS 智能筒型網(wǎng)絡(luò)攝像機(jī), 山東省濟(jì)南市甸柳莊支行 1 號柜臺,type=2,display_image=no"),
KV.of("DS-2CD6984F-IHS", " 星光級全景拼接網(wǎng)絡(luò)攝像機(jī), 山東省濟(jì)南市市中區(qū)支行 4 號柜臺,type=2,display_image=no"),
KV.of("DS-2CD7627HWD-LZS", " 全結(jié)構(gòu)化攝像機(jī), 山東省濟(jì)南市市中區(qū)支行 2 號柜臺,type=2,display_image=no"));
PCollection> notice = pipeline.apply("CreateEmails", Create.of(txtnoticelist));
PCollection> message = pipeline.apply("CreatePhones", Create.of(aimessagelist));
final TupleTag noticeTag = new TupleTag<>();
final TupleTag messageTag = new TupleTag<>();
PCollection> results = KeyedPCollectionTuple.of(noticeTag, notice).and(messageTag, message).apply(CoGroupByKey.create());
System.out.append(" 合并分組后的結(jié)果:\r");
PCollection contactLines = results.apply(ParDo.of(new DoFn, String>() {
private static final long serialVersionUID = 1L;
@ProcessElement
public void processElement(ProcessContext c) {
KV e = c.element();
String name = e.getKey();
Iterable emailsIter = e.getValue().getAll(noticeTag);
Iterable phonesIter = e.getValue().getAll(messageTag);
System.out.append("" + name + ";" + emailsIter + ";" + phonesIter + ";" + "\r");
}
}));
pipeline.run().waitUntilFinish();

2. 測試運(yùn)行結(jié)果

 

 

源碼地址： pipelineTest2_5.java

七.小結(jié)

近幾年隨著 AloT 發(fā)展得如火如荼，其落地場景也遍地開花。loT 作為 AI 落地先鋒，已經(jīng)步入線下各行各業(yè)。本文以 Beam 管道的設(shè)計切入，重點(diǎn)對 Beam 管道設(shè)計工具和源碼進(jìn)行解析，最后結(jié)合銀行金融行業(yè)對 AI 碎片化的場景進(jìn)行數(shù)據(jù)預(yù)處理的案例，幫助大家全面了解 Beam 管道。

作者介紹

張海濤，目前就職于�？低�視云基礎(chǔ)平臺，負(fù)責(zé)海康威視在全國金融行業(yè) AI 大數(shù)據(jù)落地的基礎(chǔ)架構(gòu)設(shè)計和中間件的開發(fā)，專注 AI 大數(shù)據(jù)方向。Apache Beam 中文社區(qū)發(fā)起人之一，如果想進(jìn)一步了解最新 Apache Beam 和 ClickHouse 動態(tài)和技術(shù)研究成果，請加微信 cyrjkj 入群共同研究和運(yùn)用。

標(biāo)簽： 大數(shù)據(jù)管道 機(jī)器學(xué)習(xí)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。