文 | 陳肅 DataPipeline CTO
隨着企業應用復雜性的上升和微服務架構的流行，數據正變得越來越以應用為中心。
服務之間僅在必要時以接口或者消息隊列方式進行數據交互，從而避免了構建單一數據庫集群來支撐不斷增長的業務需要。以應用為中心的數據持久化架構，在帶來可伸縮性好處的同時，也給數據的融合計算帶來了障礙。
由於數據散落在不同的數據庫、消息隊列、文件系統中，計算平台如果直接訪問這些數據，會遇到可訪問性和數據傳輸延遲等問題。在一些場景下，計算平台直接訪問應用系統數據庫會對系統吞吐造成顯著影響，通常也是不被允許的。
因此，在進行跨應用的數據融合計算時，首先需要將數據從孤立的數據源中采集出來，匯集到可被計算平台高效訪問的目的地，此過程被稱為ETL，即數據的抽取（Extract）、轉換（Transform）和加載（Load）。
ETL並不是什么新鮮事物。
該領域的傳統公司，例如Informatica，早在1993年就已經成立，並且提供了成熟的商業化解決方案。開源工具，例如Kettle、DataX等，在很多企業中也得到了廣泛的應用。
傳統上，ETL是通過批量作業完成的。即定期從數據源加載（增量）數據，按照轉換邏輯進行處理，並寫入目的地。根據業務需要和計算能力的不同，批量處理的延時通常從天到分鍾級不等。在一些應用場景下，例如電子商務網站的商品索引更新，ETL需要盡可能短的延遲，這就出現了實時ETL的需求。
在實時ETL中，數據源和數據目的地之間仿佛由管道連接在一起。數據從源端產生后，以極低的延遲被采集、加工，並寫入目的地，整個過程沒有明顯的處理批次邊界。
實時ETL，又被稱為Data Pipeline模式。
阿里提出了“數據中台”的概念。即數據被統一采集，規范數據語義和業務口徑形成企業基礎數據模型，提供統一的分析查詢和新業務的數據對接能力。
數據中台並不是新的顛覆式技術，而是一種企業數據資產管理和應用方法學，涵蓋了數據集成、數據質量管理、元數據+主數據管理、數倉建模、支持高並發訪問的數據服務接口層開發等內容。
在數據中台建設中，結合企業自身的業務需求特點，架構和功能可能各不相同，但其中一個最基本的需求是數據采集的實時性和完整性。數據從源端產生，到被采集到數據匯集層的時間要盡可能短，至少應做到秒級延遲，這樣中台的數據模型更新才可能做到近實時，構建在中台之上依賴實時數據流驅動的應用（例如商品推薦、欺詐檢測等）才能夠滿足業務的需求。
以阿里雙十一為例，在極高的並發情況下，訂單產生到大屏統計數據更新延遲不能超過5s，一般在2s內。
中台對外提供的數據應該是完整的，源端數據的Create、Update和Delete都要能夠被捕獲，不能少也不能多，即數據需要有端到端一致性的能力（Exactly Once Semantic，EOS）。
當然，EOS並非在任何業務場景下都需要，但從平台角度必須具備這種能力，並且允許用戶根據業務需求靈活開啟和關閉。
在構建實時數據集成平台時，就一些技術選型問題，建議做以下考量：
一、數據源變化捕獲
源數據變化捕獲是數據集成的起點，獲取數據源變化主要有三種方式：
-
基於日志的解析模式；
-
基於增量條件查詢模式；
-
數據源主動Push模式。
基於日志的解析模式常用於各種類型的數據庫，例如MySQL的Binlog、Oracle的Redo&Achieve Log、SQL Server Change Tracking & CDC等。
不同數據庫日志解析的原理差別很大，以MySQL Binlog模式為例，解析程序本身是一個Slave，能夠實時收到MySQL Master的數據流推送，並解析還原成DDL和DML操作。而SQL Server的CT模式下，增量是通過定期查詢Change Tracking表實現的。
基於增量條件的查詢模式不依賴於源端開啟日志記錄，但對於數據源通常有額外的格式要求。例如，數據庫表或文檔對象需要有標志更新時間的字段，這在一些業務系統中是無法滿足的。
數據源主動Push模式的常見形式為業務插碼，即應用系統通過打點或者配置切面的方式，將數據變化封裝為事件，額外發送一份給數據集成平台。這種方式一般需要對源端系統代碼進行一定程度的修改。
通常而言，基於數據庫的日志進行增量捕獲應當被優先考慮。其具備以下幾個顯著優點：
-
能夠完整獲取數據變化的操作類型，尤其是Delete操作，這是增量條件查詢模式很難做到的；
-
不依賴特別的數據字段語義，例如更新時間；
-
多數情況下具備較強的實時性。
當然，事物都具有兩面性。開啟數據庫日志通常會對源庫性能產生一定的影響，需要額外的存儲空間，甚至一些解析方法也會對源庫資源造成額外消耗。因此，實施過程中需要在DBA的配合下，根據數據庫特點和解析原理進行DB部署規划。
推薦使用數據庫的復制和災備能力，在獨立服務器對從庫進行日志解析。此外，當數據庫產生批量更新時，會在短時間內產生大量日志堆積，如果日志留存策略設置不當，容易出現數據丟失。這些都需要根據具體的業務數據增長特點，在前期做好規划，並在上線后根據業務變化定期進行評估和調整。
數據源主動push模式下，由於事件發送和業務處理很難做到事務一致性，所以當出現異常時，數據一致性就無從保證，比較適合對於數據一致性要求不高的場景，例如用戶行為分析。
二、運行環境
無論采用何種數據變化捕獲技術，程序必須在一個可靠的平台運行。該平台需要解決分布式系統的一些共性問題，主要包括：水平擴展、容錯、進度管理等。
1. 水平擴展
程序必須能夠以分布式job的形式在集群中運行，從而允許在業務增長時通過增加運行時節點的方式實現擴展。
因為在一個規模化的企業中，通常要同時運行成百上千的job。隨着業務的增長，job的數量以及job的負載還有可能持續增長。
2. 容錯
分布式運行環境的執行節點可能因為過載、網絡連通性等原因無法正常工作。
當節點出現問題時，運行環境需要能夠及時監測到，並將問題節點上的job分配給健康的節點繼續運行。
3. 進度管理
job需要記錄自身處理的進度，避免重復處理數據。另外，job會因為上下游系統的問題、網絡連通性、程序bug等各種原因異常中止，當job重啟后，必須能夠從上次記錄的正常進度位置開始處理后繼的數據。
有許多優秀的開源框架都可以滿足上述要求，包括Kafka Connect、Spark、Flink等。
Kafka Connect是一個專注數據進出Kafka的數據集成框架。Spark和Flink則更為通用，既可以用於數據集成，也適用於更加復雜的應用場景，例如機器學習的模型訓練和流式計算。
就數據集成這一應用場景而言，不同框架的概念是非常類似的。
首先，框架提供Source Connector接口封裝對數據源的訪問。應用開發者基於這一接口開發適配特定數據源的Connector，實現數據抽取邏輯和進度（offset）更新邏輯。
其次，框架提供一個分布式的Connector運行環境，處理任務的分發、容錯和進度更新等問題。
不同之處在於，Kafka Connect總是將數據抽取到Kafka，而對於Spark和Flink，Source Connector是將數據抽取到內存中構建對象，寫入目的地是由程序邏輯定義的，包括但不限於消息隊列。
但無論采用何種框架，都建議首先將數據寫入一個匯集層，通常是Kafka這樣的消息隊列。
單就數據源采集而言，Kafka Connect這樣專注於數據集成的框架是有一定優勢的，這主要體現在兩方面：
首先是Connector的豐富程度，幾乎所有較為流行的數據庫、對象存儲、文件系統都有開源的Connector實現。
尤其在數據庫的CDC方面，有Debezium這樣優秀的開源項目存在，降低了應用的成本。
其次是開發的便捷性，專有框架的設計相較於通用框架更為簡潔，開發新的Connector門檻較低。Kafka Connect的runtime實現也較為輕量，出現框架級別問題時debug也比較便捷。
盡管目前版本的Kafka Connect還不支持數據采集后進入Kafka的EOS保證，但通過對runtime的修改，利用Kafka事務消息也能夠實現這一點。相信Kafka Connect未來的版本也會很快提供官方的支持。
三、數據匯集層
當各類數據從源端抽取后，首先應當被寫入一個數據匯集層，然后再進行后繼的轉換處理，直至將最終結果寫入目的地。數據匯集層的作用主要有兩點：
首先，數據匯集層將異構的數據源數據存儲為統一的格式，並且為后繼的處理提供一致的訪問接口。這就將處理邏輯和數據源解耦開來，同時屏蔽了數據抽取過程中可能發生的異常對后繼作業的影響。
其次，數據匯集層獨立於數據源，可被多次訪問，亦可根據業務需要緩存全部或一定期限的原始數據，這為轉換分析提供了更高的靈活度。當業務需求發生變化時，無需重復讀取源端數據，直接基於數據匯集層就可以開發新的模型和應用。數據匯集層可基於任意支持海量/高可用的文件系統、數據倉庫或者消息隊列構建，常見的方案包括HDFS、Hbase、Kafka等。
針對實時ETL場景，推薦使用Kafka或類似具有海量數據持久化能力的消息隊列來做數據匯集層，這會為后繼的流式處理提供便捷。同時，利用Kafka的數據回收機制，可以根據業務需要自動保留一定時間或大小的原始數據。
四、數據轉換
數據轉換是一個業務性很強的處理步驟。
當數據進入匯集層后，一般會用於兩個典型的后繼處理場景：數倉構建和數據流服務。
數倉構建包括模型定義和預計算兩部分。數據工程師根據業務分析需要，使用星型或雪花模型設計數據倉庫結構，利用數據倉庫中間件完成模型構建和更新。
開源領域，Apache Kylin是預聚合模式OLAP代表，支持從HIVE、Kafka、HDFS等數據源加載原始表數據，並通過Spark/MR來完成CUBE構建和更新。
Druid則是另一類預聚合OLAP的代表。在Druid的表結構模型中，分為時間列、維度列和指標列，允許對任意指標列進行聚合計算而無需定義維度數量。Druid 在數據存儲時便可對數據進行聚合操作，這使得其更新延遲可以做到很低。在這些方面，Baidu開源的Palo和Druid有類似之處。
一個普遍的共識是，沒有一個OLAP引擎能同時在數據量，靈活性和性能這三個方面做到完美，用戶需要基於自己的需求進行取舍和選型。預計算模式的OLAP引擎在查詢響應時間上相較於MPP引擎（Impala、SparkSQL、Presto等）有一定優勢，但相對限制了靈活性。
如前文所述，源端采集的數據建議放入一個匯集層，優選是類似Kafka這樣的消息隊列。包括Kylin和Druid在內的數據倉庫可以直接以流式的方式消費數據進行更新。
一種常見的情形為：原始采集的數據格式、粒度不一定滿足數據倉庫中表結構的需要，而數倉提供的配置靈活度可能又不足夠。這種情況下需要在進入數倉前對數據做額外的處理。
常見的處理包括過濾、字段替換、嵌套結構一拆多、維度填充等，以上皆為無狀態的轉換。有狀態的轉換，例如SUM、COUNT等，在此過程中較少被使用，因為數倉本身就提供了這些聚合能力。
數據流服務的構建則是基於流式計算引擎，對匯集層的數據進一步加工計算，並將結果實時輸出給下游應用系統。這涉及到流式計算引擎的選擇：Spark Streaming、Flink、還是Kafka Streams？
關於三個引擎的對比，網上有很多資料，在此不再贅述。
選型過程中有幾點值得特別關注：
1. 延遲性
Spark對流的支持是MicroBatch，提供的是亞秒級的延遲，相較於Flink和Kafka Streams在實時性上要差一些。
2. 應用模式
Spark和Flink都是將作業提交到計算集群上運行，需要搭建專屬的運行環境。
Kafka Streams的作業是以普通Java程序方式運行，本質上是一個調用Kafka Streaming API的Kafka Consumer，可以方便地嵌入各種應用。
但相應的，用戶需要自己解決作業程序在不同服務器上的分發問題，例如通過K8s集群方案進行應用的容器化部署。如果使用KSQL，還需要部署KSQL的集群。
3. SQL支持
三者都提供Streaming SQL，但Flink的SQL支持要更為強大些，可以運行更加復雜的分組聚合操作。
4. EOS
Flink對於數據進出計算集群提供了框架級別的支持，這是通過結合CheckPoint機制和Sink Connector接口封裝的二階段提交協議實現的。
Kafka Streams利用Kafka事務性消息，可以實現“消費-計算-寫入Kafka“的EOS，但當結果需要輸出到Kafka以外的目的地時，還需要利用Kafka Connect的Sink Connector。
遺憾的是，Kafka Connect不提供Kafka到其它類型Sink的EOS保證，需要用戶自己實現。
Spark Streaming與Kafka Streams類似，在讀取和計算過程中可以保證EOS，但將結果輸出到外部時，依然需要額外做一些工作來確保數據一致性。常見的方式包括：利用數據庫的事務寫入機制將Offset持久化到外部、利用主鍵保證冪等寫入、參考二階段提交協議做分布式事務等。
小結
本文簡要討論了一些構建面向實時數據的集成平台在技術選型方面的考量點。
數據源變化捕獲是數據集成的起點，結合日志的解析、增量條件查詢模式和數據源主動Push模式，最終構建出一個數據匯集層。在這個階段，推薦考慮Kafka Connect這類面向數據集成的專有框架，可以有效縮短研發周期和成本。
數據匯集層建議構建在消息隊列之上，為后繼的加工處理提供便利。如果需要全量持久化長期保存，建議結合使用消息隊列和分布式文件系統分別做實時數據和全量數據的存儲。
流式處理能力是實時數據集成平台必要的組件。結合企業技術棧特點，選用包括Flink、Spark Streaming、Kafka Streams等流行的引擎在多數情況下都能夠滿足要求。
端到端數據的EOS是數據集成中的一個難題，需要用戶根據業務實際需求、數據本身的特性、目的地特點case by case去解決。