[[182474]]

本文由InfoQ中文站社區(qū)編輯足下編譯并分享

“神說(shuō)，要有光，就有了光”。——《圣經(jīng)》

1月10日，Apache軟件基金會(huì)宣布，Apache Beam成功孵化，成為該基金會(huì)的一個(gè)新的***項(xiàng)目，基于Apache V2許可證開(kāi)源。

2003年，谷歌發(fā)布了著名的大數(shù)據(jù)三篇論文，史稱(chēng)三駕馬車(chē)：Google FS、MapReduce、BigTable。雖然谷歌沒(méi)有公布這三個(gè)產(chǎn)品的源碼，但是她這三個(gè)產(chǎn)品的詳細(xì)設(shè)計(jì)論文開(kāi)啟了全球的大數(shù)據(jù)時(shí)代!從Doug Cutting大神根據(jù)谷歌的論文實(shí)現(xiàn)出Hadoop+MapReduce的雛形，到Hadoop生態(tài)圈各種衍生產(chǎn)品的蓬勃發(fā)展，再到后來(lái)的Spark、流式計(jì)算等等，所有的一切都要?dú)w功于、源自這三篇論文。可惜谷歌雖然開(kāi)啟了這個(gè)偉大的時(shí)代，卻始終僅僅滿(mǎn)足于偶爾發(fā)表一兩篇論文以強(qiáng)調(diào)自己在理論和工程上的領(lǐng)導(dǎo)地位，從來(lái)沒(méi)有親身參與進(jìn)來(lái)，尤其是沒(méi)有為開(kāi)源生態(tài)做出什么貢獻(xiàn)，因而一直沒(méi)有從大數(shù)據(jù)市場(chǎng)獲得什么實(shí)在的好處。

痛定思痛，谷歌開(kāi)始走開(kāi)源之路，將自己的標(biāo)準(zhǔn)推廣給社區(qū)。從眾所周知的Kubernetes，到2016年2月谷歌高調(diào)宣布將Apache Beam(原名Google DataFlow)貢獻(xiàn)給Apache基金會(huì)孵化，再到最近大熱的Tensorflow等等，動(dòng)作不斷。Apache Beam被認(rèn)為是繼MapReduce，GFS和BigQuery等之后，谷歌在大數(shù)據(jù)處理領(lǐng)域?qū)﹂_(kāi)源社區(qū)的又一個(gè)非常大的貢獻(xiàn)。

也就是說(shuō)，在大數(shù)據(jù)處理的世界里，谷歌一直在內(nèi)部閉源，開(kāi)發(fā)并使用著B(niǎo)igTable、Spanner、Millwheel等讓大家久聞大名而又無(wú)緣一見(jiàn)的產(chǎn)品，開(kāi)源世界演進(jìn)出了Hadoop、Spark、Apache Flink等產(chǎn)品，現(xiàn)在他們終于殊途同歸，走到一起來(lái)了。 

為什么要推出開(kāi)源的Apache Beam?

Apache Beam的主要負(fù)責(zé)人Tyler Akidau在他的博客中提到他們做這件事的理念是：

要為這個(gè)世界貢獻(xiàn)一個(gè)容易使用而又強(qiáng)大的模型，用于大數(shù)據(jù)的并行處理，同時(shí)適用于流式處理和批量處理，而且在各種不同平臺(tái)上還可以移植。

那這一次為什么不是又酷酷的發(fā)表一篇論文，然后退居一旁靜靜的觀察呢?為什么要聯(lián)合一眾伙伴為大家直接提供可以運(yùn)行的代碼了呢?原因主要有兩點(diǎn)：

• 盡管在過(guò)去谷歌一直是閉源的，但在為云客戶(hù)服務(wù)的過(guò)程中，谷歌已經(jīng)認(rèn)識(shí)到了開(kāi)源軟件的的巨大價(jià)值，比如基于谷歌三篇論文產(chǎn)生的Hadoop社區(qū)就是一個(gè)非常好的例子。思想上的轉(zhuǎn)變使Apache Beam的誕生成為可能;
• 就Beam這個(gè)項(xiàng)目而言，要成功的必要條件之一是，必須有已經(jīng)開(kāi)源的Runner為Beam模型提供充分的支持，這樣它才會(huì)在自建云和非谷歌云的場(chǎng)景下成為一個(gè)非常有競(jìng)爭(zhēng)力的備選方案。去年Apache Flink在他們的系統(tǒng)內(nèi)采用了Beam模型，這一條件也得到了滿(mǎn)足;

無(wú)利不起早，谷歌這樣做也是有著直接商業(yè)動(dòng)機(jī)的，就是希望能有盡可能多的Apache Beam數(shù)據(jù)處理流水線(xiàn)可以運(yùn)行在谷歌的Cloud Dataflow上，別忘了這是Apache Beam的原型。進(jìn)一步說(shuō)，采用開(kāi)源的方式來(lái)引導(dǎo)這件事，也是有許多直接好處的：

• 支持Apache Beam的Runner越多，它作為一個(gè)平臺(tái)的吸引力就越大;
• 使用Apache Beam的用戶(hù)越多，想在谷歌云平臺(tái)上運(yùn)行Apache Beam的用戶(hù)也就越多;
• 開(kāi)發(fā)Apache Beam過(guò)程中吸引到的伙伴越多，那對(duì)這樣的數(shù)據(jù)處理模型的推廣就越有利;

而且，好處也不會(huì)全都?xì)w于谷歌，Apache Beam項(xiàng)目中的所有參與方都會(huì)受益。如果在構(gòu)建數(shù)據(jù)處理流水線(xiàn)時(shí)存在著這樣一個(gè)可移植的抽象層，那就會(huì)更容易出現(xiàn)新的Runner，它們可以專(zhuān)注于技術(shù)創(chuàng)新，提供更高的性能、更好的可靠性、更方便的運(yùn)維管理等。換句話(huà)說(shuō)，消除了對(duì)API的鎖定，就解放了處理引擎，會(huì)導(dǎo)致更多產(chǎn)品之間的競(jìng)爭(zhēng)，從而最終對(duì)整個(gè)行業(yè)起到良性的促進(jìn)作用。

谷歌堅(jiān)信Apache Beam就是數(shù)據(jù)批量處理和流式處理的未來(lái)。這么做會(huì)為各種不同的Runner營(yíng)造一個(gè)健康的生態(tài)系統(tǒng)，讓它們之間相互競(jìng)爭(zhēng)，而***可以讓用戶(hù)得到實(shí)在的好處。

Apache Beam是什么?

要說(shuō)Apache Beam，先要說(shuō)說(shuō)谷歌Cloud Dataflow。Dataflow是一種原生的谷歌云數(shù)據(jù)處理服務(wù)，是一種構(gòu)建、管理和優(yōu)化復(fù)雜數(shù)據(jù)流水線(xiàn)的方法，用于構(gòu)建移動(dòng)應(yīng)用、調(diào)試、追蹤和監(jiān)控產(chǎn)品級(jí)云應(yīng)用。它采用了谷歌內(nèi)部的技術(shù)Flume和MillWhell，其中Flume用于數(shù)據(jù)的高效并行化處理，而MillWhell則用于互聯(lián)網(wǎng)級(jí)別的帶有很好容錯(cuò)機(jī)制的流處理。該技術(shù)提供了簡(jiǎn)單的編程模型，可用于批處理和流式數(shù)據(jù)的處理任務(wù)。她提供的數(shù)據(jù)流管理服務(wù)可控制數(shù)據(jù)處理作業(yè)的執(zhí)行，數(shù)據(jù)處理作業(yè)可使用DataFlow SDK創(chuàng)建。 

Apache Beam本身不是一個(gè)流式處理平臺(tái)，而是一個(gè)統(tǒng)一的編程框架，它提供了開(kāi)源的、統(tǒng)一的編程模型，幫助你創(chuàng)建自己的數(shù)據(jù)處理流水線(xiàn)，實(shí)現(xiàn)可以運(yùn)行在任意執(zhí)行引擎之上批處理和流式處理任務(wù)。Beam對(duì)流式計(jì)算場(chǎng)景中的所有問(wèn)題重新做了一次歸納，然后針對(duì)這些問(wèn)題提出了幾種不同的解決模型，然后再把這些模型通過(guò)一種統(tǒng)一的語(yǔ)言給實(shí)現(xiàn)出來(lái)，最終這些Beam程序可以運(yùn)行在任何一個(gè)計(jì)算平臺(tái)上(只要相應(yīng)平臺(tái)——即Runner實(shí)現(xiàn)了對(duì)Beam的支持)。它的特點(diǎn)有：

• 統(tǒng)一的：對(duì)于批處理和流式處理，使用單一的編程模型;
• 可移植的：可以支持多種執(zhí)行環(huán)境，包括Apache Apex、Apache Flink、Apache Spark和谷歌Cloud Dataflow等;
• 可擴(kuò)展的：可以實(shí)現(xiàn)和分享更多的新SDK、IO連接器、轉(zhuǎn)換操作庫(kù)等;

Beam特別適合應(yīng)用于并行數(shù)據(jù)處理任務(wù)，只要可以將要處理的數(shù)據(jù)集分解成許多相互獨(dú)立而又可以并行處理的小集合就可以了。Beam也可以用于ETL任務(wù)，或者單純的數(shù)據(jù)整合。這些任務(wù)主要就是把數(shù)據(jù)在不同的存儲(chǔ)介質(zhì)或者數(shù)據(jù)倉(cāng)庫(kù)之間移動(dòng)，將數(shù)據(jù)轉(zhuǎn)換成希望的格式，或者將數(shù)據(jù)導(dǎo)入一個(gè)新系統(tǒng)。

Beam主要包含兩個(gè)關(guān)鍵的部分：

• Beam SDK

Beam SDK提供一個(gè)統(tǒng)一的編程接口給到上層應(yīng)用的開(kāi)發(fā)者，開(kāi)發(fā)者不需要了解底層的具體的大數(shù)據(jù)平臺(tái)的開(kāi)發(fā)接口是什么，直接通過(guò)Beam SDK的接口，就可以開(kāi)發(fā)數(shù)據(jù)處理的加工流程，不管輸入是用于批處理的有限數(shù)據(jù)集，還是流式的***數(shù)據(jù)集。對(duì)于有限或***的輸入數(shù)據(jù)，Beam SDK都使用相同的類(lèi)來(lái)表現(xiàn)，并且使用相同的轉(zhuǎn)換操作進(jìn)行處理。Beam SDK可以有不同編程語(yǔ)言的實(shí)現(xiàn)，目前已經(jīng)完整地提供了Java，python的SDK還在開(kāi)發(fā)過(guò)程中，相信未來(lái)會(huì)有更多不同的語(yǔ)言的SDK會(huì)發(fā)布出來(lái)。

• Beam Pipeline Runner

Beam Pipeline Runner將用戶(hù)用Beam模型定義開(kāi)發(fā)的處理流程翻譯成底層的分布式數(shù)據(jù)處理平臺(tái)支持的運(yùn)行時(shí)環(huán)境。在運(yùn)行Beam程序時(shí)，需要指明底層的正確Runner類(lèi)型。針對(duì)不同的大數(shù)據(jù)平臺(tái)，會(huì)有不同的Runner。目前Flink、Spark、Apex以及谷歌的Cloud DataFlow都有支持Beam的Runner。

需要注意的是，雖然Apache Beam社區(qū)非常希望所有的Beam執(zhí)行引擎都能夠支持Beam SDK定義的功能全集，但是在實(shí)際實(shí)現(xiàn)中可能并不一定。例如，基于MapReduce的Runner顯然很難實(shí)現(xiàn)和流處理相關(guān)的功能特性。就目前狀態(tài)而言，對(duì)Beam模型支持***的就是運(yùn)行于谷歌云平臺(tái)之上的Cloud Dataflow，以及可以用于自建或部署在非谷歌云之上的Apache Flink。當(dāng)然，其它的Runner也正在迎頭趕上，整個(gè)行業(yè)也在朝著支持Beam模型的方向發(fā)展。

那大家可以怎樣與Beam做親密接觸呢? 

如上圖所示，主要有三個(gè)方面：

• 數(shù)據(jù)處理：直接使用已有的自己熟悉語(yǔ)言的SDK，根據(jù)Beam模型去定義并實(shí)現(xiàn)自己的數(shù)據(jù)處理流程;
• SDK實(shí)現(xiàn)：用新的編程語(yǔ)言去根據(jù)Beam概念實(shí)現(xiàn)SDK，這樣大家以后在編程語(yǔ)言方面就可以有更多選擇了;
• Runner實(shí)現(xiàn)：將已有的分布式數(shù)據(jù)處理平臺(tái)作為一種新的Runner，接入Beam模型;

Beam是怎么做的?

在任何一個(gè)設(shè)計(jì)開(kāi)始之前，都先要確定問(wèn)題，Beam也不例外。

1. 數(shù)據(jù)。分布式數(shù)據(jù)處理要處理的數(shù)據(jù)類(lèi)型一般可以分為兩類(lèi)，有限的數(shù)據(jù)集和***的數(shù)據(jù)流。有限的數(shù)據(jù)集，比如一個(gè)HDFS中的文件，一個(gè)HBase表等，特點(diǎn)是數(shù)據(jù)提前已經(jīng)存在，一般也已經(jīng)持久化，不會(huì)突然消失，不會(huì)再改變。而***的數(shù)據(jù)流，比如kafka中流過(guò)來(lái)的系統(tǒng)日志流，或是從Twitter API拿到的Twitter流等等，這類(lèi)數(shù)據(jù)的特點(diǎn)是，數(shù)據(jù)動(dòng)態(tài)流入，無(wú)窮無(wú)盡，無(wú)法全部持久化。一般來(lái)說(shuō)，批處理框架的設(shè)計(jì)目標(biāo)是用來(lái)處理有限的數(shù)據(jù)集，流處理框架的設(shè)計(jì)目標(biāo)是用來(lái)處理***的數(shù)據(jù)流。有限的數(shù)據(jù)集可以看做是***的數(shù)據(jù)流的一種特例，但是從數(shù)據(jù)處理邏輯的角度，這兩者并無(wú)不同之處。
2. 時(shí)間。Process Time是指數(shù)據(jù)進(jìn)入分布式處理框架的時(shí)間，而Event-Time則是指數(shù)據(jù)產(chǎn)生的時(shí)間。這兩個(gè)時(shí)間通常是不同的，例如，對(duì)于一個(gè)處理微博數(shù)據(jù)的流計(jì)算任務(wù)，一條2016-06-01-12:00:00發(fā)表的微博經(jīng)過(guò)網(wǎng)絡(luò)傳輸?shù)妊舆t可能在2016-06-01-12:01:30才進(jìn)入到流處理系統(tǒng)中。批處理任務(wù)通常進(jìn)行全量的數(shù)據(jù)計(jì)算，較少關(guān)注數(shù)據(jù)的時(shí)間屬性，但是對(duì)于流處理任務(wù)來(lái)說(shuō)，由于數(shù)據(jù)流是無(wú)情無(wú)盡的，無(wú)法進(jìn)行全量的計(jì)算，通常是對(duì)某個(gè)窗口中得數(shù)據(jù)進(jìn)行計(jì)算，對(duì)于大部分的流處理任務(wù)來(lái)說(shuō)，按照時(shí)間進(jìn)行窗口劃分，可能是最常見(jiàn)的需求。
3. 亂序。對(duì)于流處理框架處理的數(shù)據(jù)流來(lái)說(shuō)，其數(shù)據(jù)的到達(dá)順序可能并不嚴(yán)格按照Event-Time的時(shí)間順序。如果基于Process Time定義時(shí)間窗口，數(shù)據(jù)到達(dá)的順序就是數(shù)據(jù)的順序，因此不存在亂序問(wèn)題。但是對(duì)于基于Event Time定義的時(shí)間窗口來(lái)說(shuō)，可能存在時(shí)間靠前的消息在時(shí)間靠后的消息之后到達(dá)的情況，這在分布式的數(shù)據(jù)源中可能非常常見(jiàn)。對(duì)于這種情況，如何確定遲到數(shù)據(jù)，以及對(duì)于遲到數(shù)據(jù)如何處理通常是很棘手的問(wèn)題。

Beam模型處理的目標(biāo)數(shù)據(jù)是***的時(shí)間亂序數(shù)據(jù)流，不考慮時(shí)間順序或是有限的數(shù)據(jù)集可看做是***亂序數(shù)據(jù)流的一個(gè)特例。 

如上圖，其中虛線(xiàn)是最理想的，表示處理時(shí)間和事件時(shí)間是相同的，紅線(xiàn)是實(shí)際上的線(xiàn)，也叫水位線(xiàn)(Watermark)，它一般是通過(guò)啟發(fā)式算法算出來(lái)的。

接下來(lái)從問(wèn)題中抽象出四個(gè)具體的問(wèn)題：

A：What are you computing，對(duì)數(shù)據(jù)的處理是哪種類(lèi)型，數(shù)據(jù)轉(zhuǎn)換、聚合或者是兩者都有。例如，Sum、Join或是機(jī)器學(xué)習(xí)中訓(xùn)練學(xué)習(xí)模型等。在Beam SDK中由Pipeline中的操作符指定。如圖： 

B：Where in event time，數(shù)據(jù)在什么范圍中計(jì)算?例如，基于Process-Time的時(shí)間窗口?基于Event-Time的時(shí)間窗口?滑動(dòng)窗口等等。在Beam SDK中由Pipeline中的窗口指定： 

C：When in processing time，何時(shí)將計(jì)算結(jié)果輸出?在這里引入了一個(gè)Trigger機(jī)制，Trigger決定何時(shí)將計(jì)算結(jié)果發(fā)射出去，發(fā)射太早會(huì)丟失一部分?jǐn)?shù)據(jù)，喪失精確性，發(fā)射太晚會(huì)導(dǎo)致延遲變長(zhǎng)，而且會(huì)囤積大量數(shù)據(jù)，何時(shí)Trigger是由水位線(xiàn)來(lái)決定的，在Beam SDK中由Pipeline中的水位線(xiàn)和觸發(fā)器指定。 

D：How do refinements relate，遲到數(shù)據(jù)如何處理?例如，將遲到數(shù)據(jù)計(jì)算增量結(jié)果輸出，或是將遲到數(shù)據(jù)計(jì)算結(jié)果和窗口內(nèi)數(shù)據(jù)計(jì)算結(jié)果合并成全量結(jié)果輸出。在Beam SDK中由Accumulation指定。 

Beam模型將”WWWH“四個(gè)維度抽象出來(lái)組成了Beam SDK，用戶(hù)在基于Beam SDK構(gòu)建數(shù)據(jù)處理業(yè)務(wù)邏輯時(shí)，每一步只需要根據(jù)業(yè)務(wù)需求按照這四個(gè)維度調(diào)用具體的API即可生成分布式數(shù)據(jù)處理Pipeline，并提交到具體執(zhí)行引擎上執(zhí)行。“WWWH”四個(gè)維度的抽象僅僅關(guān)注業(yè)務(wù)邏輯本身，和分布式任務(wù)如何執(zhí)行沒(méi)有任何關(guān)系。

友商的看法

隨著分布式數(shù)據(jù)處理不斷發(fā)展，新的分布式數(shù)據(jù)處理技術(shù)也不斷被提出，業(yè)界涌現(xiàn)出了越來(lái)越多的分布式數(shù)據(jù)處理框架，從最早的Hadoop MapReduce，到Apache Spark，Apache Storm，以及更近的Apache Flink，Apache Apex等。新的分布式處理框架可能帶來(lái)的更高的性能，更強(qiáng)大的功能，更低的延遲等，但用戶(hù)切換到新的分布式處理框架的代價(jià)也非常大：需要學(xué)習(xí)一個(gè)新的數(shù)據(jù)處理框架，并重寫(xiě)所有的業(yè)務(wù)邏輯。解決這個(gè)問(wèn)題的思路包括兩個(gè)部分，首先，需要一個(gè)編程范式，能夠統(tǒng)一，規(guī)范分布式數(shù)據(jù)處理的需求，例如，統(tǒng)一批處理和流處理的需求。其次，生成的分布式數(shù)據(jù)處理任務(wù)應(yīng)該能夠在各個(gè)分布式執(zhí)行引擎上執(zhí)行，用戶(hù)可以自由切換分布式數(shù)據(jù)處理任務(wù)的執(zhí)行引擎與執(zhí)行環(huán)境。Apache Beam正是為了解決以上問(wèn)題而提出的。

如Apache Beam項(xiàng)目的主要推動(dòng)者Tyler Akidau所說(shuō)：

“為了讓Apache Beam能成功地完成移植，我們需要至少有一個(gè)在部署自建云或非谷歌云時(shí)，可以與谷歌Cloud Dataflow相比具備足夠競(jìng)爭(zhēng)力的Runner。如Beam能力矩陣所示，F(xiàn)link滿(mǎn)足我們的要求。有了Flink，Beam已經(jīng)在業(yè)界內(nèi)成了一個(gè)真正有競(jìng)爭(zhēng)力的平臺(tái)。”

對(duì)此，Data Artisan的Kostas Tzoumas在他的博客中說(shuō)：

“在谷歌將他們的Dataflow SDK和Runner捐獻(xiàn)給Apache孵化器成為Apache Beam項(xiàng)目時(shí)，谷歌希望我們能幫忙完成Flink Runner，并且成為新項(xiàng)目的代碼提交者和PMC成員。我們決定全力支持，因?yàn)槲覀冋J(rèn)為：1、對(duì)于流處理和批處理來(lái)說(shuō)Beam模型都是未來(lái)的參考架構(gòu);2、Flink正是一個(gè)執(zhí)行這樣數(shù)據(jù)處理的平臺(tái)。在Beam成形之后，現(xiàn)在Flink已經(jīng)成了谷歌云之外運(yùn)行Beam程序的***平臺(tái)。

我們堅(jiān)信Beam模型是進(jìn)行數(shù)據(jù)流處理和批處理的***編程模型。我們鼓勵(lì)用戶(hù)們?cè)趯?shí)現(xiàn)新程序時(shí)采用這個(gè)模型，用Beam API或者Flink DataStream API都行。”

目前主流流數(shù)據(jù)處理框架Flink、Spark、Apex以及谷歌的Cloud DataFlow等都有了支持Beam的Runner。

結(jié)論

“在谷歌公司里已經(jīng)沒(méi)人再使用MapReduce了”!谷歌云的主要負(fù)責(zé)人Mete Atamel如是說(shuō)。谷歌堅(jiān)信Apache Beam就是數(shù)據(jù)批處理和流處理的未來(lái)。Apache Beam的模型對(duì)***亂序數(shù)據(jù)流的數(shù)據(jù)處理進(jìn)行了非常優(yōu)雅的抽象，“WWWH”四個(gè)維度對(duì)數(shù)據(jù)處理的描述非常清晰與合理，Beam模型在統(tǒng)一了對(duì)***數(shù)據(jù)流和有限數(shù)據(jù)集的處理模式的同時(shí)，也明確了對(duì)***數(shù)據(jù)流的數(shù)據(jù)處理方式的編程范式，擴(kuò)大了流處理系統(tǒng)可應(yīng)用的業(yè)務(wù)范圍。隨著Apache Beam的成功孵化，隨著越來(lái)越多的編程語(yǔ)言可用、越來(lái)越多的分布式數(shù)據(jù)處理平臺(tái)支持Beam模型，我們的確可以盡情暢想美好的未來(lái)。

QCon北京2017將于4月16日~18日在北京·國(guó)家會(huì)議中心舉行，精心設(shè)計(jì)了互聯(lián)網(wǎng)廣告系統(tǒng)實(shí)踐、大數(shù)據(jù)實(shí)時(shí)計(jì)算與流處理和金融科技轉(zhuǎn)型與未來(lái)等30來(lái)個(gè)專(zhuān)題，將邀請(qǐng)來(lái)自Google、Facebook、阿里巴巴、騰訊、百度、美團(tuán)點(diǎn)評(píng)、愛(ài)奇藝等典型互聯(lián)網(wǎng)公司的技術(shù)專(zhuān)家，分享技術(shù)領(lǐng)域***成果。