導讀：阿里云DataWorks數據集成是DataX團隊在阿里云上的商業化產品，致力于提供復雜網絡環境下、豐富的異構數據源之間高速穩定的數據移動能力，以及繁雜業務背景下的數據同步解決方案。目前已經支持云上近3000家客戶，單日同步數據超過3萬億條。DataWorks數據集成目前支持離線50+種數據源，可以進行整庫遷移、批量上云、增量同步、分庫分表等各類同步解決方案。2020年上線實時同步能力，支持10+種數據源的讀寫任意組合，提供MySQL，Oracle等多種數據源到阿里云MaxCompute，Hologres等大數據引擎的一鍵全增量同步解決方案。今天跟大家交流的內容是數據集成DataX的架構，以及在使用數據集成這款產品過程中遇到的一些問題和實踐經驗。

今天的介紹會圍繞下面幾點展開：

為什么需要數據集成

阿里云數據集成大事記

阿里云數據集成產品定位和業務支撐情況

阿里云數據集成DataX設計核心思路

阿里云數據集成DataX架構

阿里云數據集成數據同步-核心亮點

阿里云數據集成解決方案系統

阿里云 DataWorks & 數據集成關系

智能實時數倉解決方案示例

01為什么需要數據集成

1. 數據集成的應用場景

首先為什么需要數據集成，在大數據場景下，數據集成主要是解決哪些核心問題，我將其分為以下四個場景：

第一個場景就是搬站上云。搬站上云主要用來對接業務數據庫，完成數倉初始化，比如用戶需要把自己的云下數據庫快速安全地遷移到云上存儲并做進一步的業務分析，如線下MySQL、Oracle到云上MaxCompute或者是Hadoop中Hive上，這既包含了一次性全量數據遷移，還包含持續增量數據遷移。

第二個場景是構建實時數倉，做流式的數據匯聚。主要包含應用產生的日志數據，應用產生的消息數據，和應用對接數據庫產生的日志數據（數據庫的Binlog或者歸檔日志）。

第三個場景是平臺融和，完成云上各個產品之間的數據同步和交換。既包含跨云平臺之間的數據同步，比如從AWS、GCP到阿里云；還包含云平臺內部之間的數據交換，比如阿里云內部RDS、MaxCompute、OSS、PolarDB、AnalyticDB、Datahub等之間的數據同步。

第四個場景是容災備份。當數據分為冷熱數據時，需要把冷數據放到歸檔數據中，極大節約成本。在數倉場景中，也可以把云上計算好的一些結果數據回流到數據庫中，服務于在線數據應用。

以上是數據集成主要支撐的四個典型的數據場景。

2. 數據集成的角色和地位

數據集成在整個數倉建設中的角色和地位是什么呢？這里我們將數倉開發簡單概括為6個模塊或步驟，第一個是數據源，里面包含了各種關系型數據庫、文件存儲、大數據存儲、消息隊列等。這些異構的數據源可以通過統一的數據集成平臺來將異構網絡的數據統一抽取到數據倉庫平臺，在數據倉庫平臺中完成數據匯聚，進行進一步的數據分析。分析后的結果數據仍可以通過數據集成回流到在線數據庫，為在線業務、在線應用提供數據查詢。除此之外，還可以利用一個承上啟下的產品-數據服務，對接數據應用，制作相關報表、大屏、和應用等。

綜上所述，我們知道數據集成在整個數據倉庫開發建設過程中起到了非常關鍵的作用，只有通過數據集成將異構數據源統一匯總到數據倉庫中，才有后序的數據分析、數據服務、數據應用等。

02阿里云數據集成大事記

阿里云數據集成正是為了解決數據倉庫中這些經典問題而出現的，它是一款有歷史淵源的產品。2011年伴隨著阿里巴巴數據平臺事業部成立，當時推出了DataX的1.0和2.0版本，當時大家可能對于這款產品還沒有太多了解。在2014年，伴隨著DataX 3.0版本的發布，我們將服務于內部的數據集成產品推向了阿里云（公有云、專有云），開始正式的對外提供服務。并且于2016年，將數據同步引擎DataX開源到社區，助力大數據ETL生態建設。在2018年DataX有了一個比較大的架構升級，將公有云、專有云、阿里內部功能統一建立全新的Data Integration服務，進一步優化了開發效率，節約了運維成本。在2019年，將數據集成能力進行了商業化。在2020年，發布了數據集成的實時同步能力，可以支持數據實時傳輸，并且提供了一些實時同步場景解決方案。

總之，阿里云數據集成是大數據平臺上下云的核心樞紐，可以將不同業務系統中數據相互打通，實現數據自由離線或實時的流動。其致力于提供復雜網絡環境下，豐富的異構數據源之間高速穩定的數據移動能力，以及在繁雜業務背景下，提供數據同步解決方案。

03阿里云數據集成產品定位和業務支撐情況1. 阿里云數據集成產品定位

圖中是阿里云數據集成產品的核心功能特性列表，大家在自己設計或選型ETL產品的時候，可以著重了解這些點，做選型參考。

首先，離線和實時全覆蓋，這款產品既支持傳統的離線同步，也支持及時高效的實時同步；

第二點，針對各種復雜網絡均做了相應的解決方案和產品化能力，無論數據源在公網、IDC還是VPC等，數據集成都有對應的解決方案；

第三點，因為DataX是一個云上產品，同時需要具備一些安全的管控策略，我們將開發和生產環境隔離，進行數據源權限安全控制；

第四點，得益于這種可擴展性架構，DataX可以支持繁多的異構數據源，離線支持50+種數據源，實時支持10+種數據源讀寫任意組合，其中涵蓋了關系型數據庫、MPP、NoSQL、文件儲存、消息流等各大種類；

第五點，在大數據場景中往往存在特定場景的通用需求，比如整庫遷移、批量上云、增量同步、分庫分表、一鍵實時全增量，我們將這些通用需求抽象為數據集成的解決方案，進一步降低用戶使用數據集成門檻；

最后我們擁有一個非常完備的運維監控體系，能夠進行流量控制、臟數據控制，能夠對任務執行資源組進行監控，另外還可以設置任務告警，支持電話、短信、郵件、釘釘等。

2. 阿里云數據集成業務支撐情況

首先該產品現在擁有數千用戶、數萬名開發者，每天同步的數據量在PB級別，這些數據是由數千萬個任務調度完成，而且該產品已經做到了全球地域部署覆蓋，目前已經涵蓋了政府、金融、保險、能源、電力、制造、互聯網、零售等多個行業，幾乎每一個行業都有典型的客戶和場景。

04阿里云數據集成DataX設計核心思路

1. 核心設計思路

左側圖是傳統的數據ETL工具制作的方法，是一種網狀模型，每種儲存之間如果交換數據，需分別設計一套數據傳輸軟件。比如在MySQL和FTP之間交換數據，需要實現一個MySQL到FTP的數據同步工具，在MySQL和HDFS之間交換數據，還要實現另一個同步工具，這種方法擴展性非常差。

針對這種場景，DataX做了一個非常核心的抽象，就是右側圖中藍色的線性模型，每一種數據源統一通過DataX框架做數據交換，當增加一種新的數據源類型同步插件，便天然的具備和DataX已有的其他數據源通道進行數據交換的能力。

下面著重介紹一下具體的實現思路，首先每一種數據存儲，比如MySQL，會有一個MySQL的Reader插件和一個Writer插件，Reader插件主要是用來讀取MySQL中的數據，讀取的模式可以是離線的同步，也可以是實時的Binlog同步，Writer插件主要是用來寫入數據。DataX做了兩個核心抽象，第一個是接口抽象，不論Reader還是Writer插件均有一套接口規范，只要符合接口規范，就可以納入平臺調度體系，被平臺調度和使用。另外一個抽象是數據類型規范，每增加一種新的數據源，只要符合數據交換規范，就可以和平臺中已經支持的其他數據源進行交換，以上就是DataX極具擴展性的原因。我們在2016年將數據同步核心同步引擎的一部分DataX開源到了社區，地址為：

https://github.com/alibaba/DataX?。

DataX是數據集成的一個核心同步引擎，是我們同步能力的一部分，今天除了要介紹一些開源的架構之外，還會介紹一些商業版，還沒有完全開源的部分功能的核心架構。

2. 離線數據同步原理

首先是離線數據的同步原理，每一個數據同步作業都是一個DataX程序進程，DataX可以連接數據源頭讀取數據，讀取到數據以后寫入到目標端。數據源頭的數據讀取可以基于數據庫的JDBC協議，這些在一些關系型數據庫或者一些MPP數據庫中已經非常常見了；如果源數據系統是Kafka，MQ消息隊列等儲存形式的話，可以使用數據源對應的SDK進行讀寫。一個任務在讀取的過程中，會發生切片、分發、執行，Reader和Writer之間是一個生產者消費者模型，Reader在讀取數據以后會將數據發送到中間的緩存隊列，Writer從中間緩存隊列捕獲數據，并最終由Writer寫入到目標儲存里面。

3. 實時數據同步原理

其次，介紹一下DataX實時數據同步原理。數據來源主要分為兩類，關系型數據庫和實時消息流。關系型數據庫可以有對應日志，比如MySQL對應的Binlog日志，Oracle的歸檔日志，或者數據變更捕獲cdc。這些日志會被日志抽取捕獲到，捕獲方式有API和SDK兩種方式，然后框架會將捕獲到的內容做消息分解處理，分解為insert、update、delete、alter等事件，然后并行地將其存放。我們支持關系型數據庫、大數據、實時消息流等多種目標存儲方式，如果數據源就是實時消息流的話，我們可以使用消息訂閱模式訂閱消息，并且將消息同樣做分解，最終重放到目標端。其中第一類關系型數據庫復雜度還是比較高的，因為包含了數據的增刪改和DDL操作，實時消息流則主要是一些追加數據。

05阿里云數據集成DataX架構1. DataX架構模式?

① Standalone模式

在了解基礎原理之后，向大家介紹一下阿里云數據集成DataX的架構設計，其中最經典的是大家比較熟悉的Standalone模式，Github上開源的即為這種模式。

Standalone模式將整個過程抽象為3個部分：

第一部分是framework框架，框架主要用來提供一些核心的、共性的功能，比如數據的速度控制、臟數據、任務執行指標的收集、匯總和上報，除此之外，框架還具有一定的調度能力；

第二部分是Reader插件部分，Reader插件主要是用來對接數據源頭，將源頭數據讀入其中；

第三部分是Writer寫插件部分，每一個目標存儲類型都對應一個Writer插件，可以將源頭投遞過來的數據，寫入到不同的目標儲存里面。

上面就是一個非常簡單的模型，我們將一個數據同步作業稱之為一個job，為了最大化的提高任務執行速度和效率，我們會將一個job拆分成多個task任務，每一個task都是一個完整數據同步作業的一個工作子任務，用來完成對應工作期間數據傳輸和投遞，task任務可以被我們調度框架進行調度執行。task是分組的，即TaskGroup。一個TaskGroup可以有多個task子任務，TaskGroup會在worker節點中執行和處理，worker也是單機版本的，這也是Github中開源的經典模式。

② Distribute模式

除了Standalone模式以外，我們在商業版本里面還有一些更強大的能力，首先是分布式模式，剛剛我們看到一個數據傳輸作業可以被分為多個task子任務，task又可以分組成TaskGroup，這些task會被并發執行。我們將前面講的調度部分換成分布式調度模型，我們可以將不同的TaskGroup分發到不同的worker節點上面，就可以突破系統的單機瓶頸，在面對海量數據同步的時候，極大地提升數據傳輸效率。分布式執行集群也可以做到線性擴展，只要數據系統吞吐可以保障，數據同步作業可以隨著worker節點的增加而線性增加，并且我們還做了一些異常節點的隔離等對應的一些分布式機制。

③ On Hadoop模式

第三個模式是On Hadoop模式，也是商業化版本的一個能力。當用戶已經擁有一個Hadoop執行集群，我們可以將DataX數據傳輸作業部署在已有的Hadoop集群里面，Hadoop中常見的編程模型是MapReduce，我們可以將DataX拆分的task寄宿在mapper節點和reducer節點中，通過Yarn進行統一調度和管理，通過這種方式我們可以復用已有的Hadoop計算和執行能力。這種模式和開源的Sqoop框架是有一點類似的。

2. 實時同步CheckPoint機制

下面介紹阿里云數據集成DataX關于實時同步的checkpoint機制。checkpoint機制可以保證實時數據傳輸的穩定和斷點續傳的能力。

如果你對Flink特別了解的話，這張圖會非常熟悉。Flink是阿里團隊開源出去的另外一個非常重要的實時計算引擎，DataX框架也借助了Flink的checkpoint機制，比如Flink會定期發送一些barrier事件和消息。我們的Reader Task其實是source，收到barrier以后，會產生snapShotState，并且barrier會傳遞到Transformer Task，Transformer Task可以用來做數據的轉換，Transformer Task收到barrier以后，barrier進一步傳遞到Writer Task，這個時候的Writer Task其實就是sink，收到barrier之后，會再做一次snapShotState，Writer Task會將我們的數據flush到目標儲存。我們會跟蹤barrier進度情況，并且根據barrier進度情況，把數據流消費的點位cache緩存下來，并且可以進行持久化存儲。當任務出現異常或者進程退出的時候，我們可以繼續從上一個cache點繼續消費數據，可以保證數據不會被丟棄，不過數據可能會有部分的重復，一般后序的計算引擎可以處理這種情況（冪等寫出，最終一致）。

06阿里云數據集成數據同步-核心亮點

1. 阿里云數據集成離線同步-核心亮點

概況一下阿里云數據集成離線同步的核心亮點。主要分為以下四個部分：

第一部分是支持多種類的數據源，DataX支持50+常見數據源，涵蓋各種關系型數據庫、文件系統、大數據系統、消息系統；

第二部分是解決方案系統，我們為一些數據傳輸經典問題準備了對應的解決方案，比如支持全量和增量的數據同步，支持整庫、批量數據同步、支持分庫分表，我們將這些瑣碎的功能整合成了產品化的解決方案，直接通過界面操作即可完整復雜的數據傳輸過程；

第三部分是精細化權限管控能力，可以對數據源權限進行安全控制，并且隔離開發和生產環境；

第四部分DataX支持復雜調度，數據集成與DataWorks深度融合，利用DataWorks強大的調度能力調度我們的數據傳輸任務。

2. 阿里云數據集成實時同步-核心亮點

再概況下阿里云數據集成實時同步的核心亮點。

DataX是借助插件化機制，對新的數據源支持擴展能力強。

DataX支持豐富多樣的數據源，支持星型鏈路組合，任何一種輸入源都可以和任何一種輸出源搭配組成同步鏈路。

DataX支持斷點續傳，可以實時讀取MySQL、Oracle、SQLSever、OceanBase、Kafka、LogHub、DataHub、PolarDB等的數據，可以將數據實時寫入到MaxCompute、Hologres、Datahub、Kafka、ElasticSearch等儲存系統。

DataX天然具有云原生基因，和阿里云產品融合度非常高。

DataX可以輕松監控運維告警，提供運維大盤、監控報警、FailOver等運維能力，可以監控業務延遲、Failover、臟數據、心跳檢查、失敗信息，并且支持郵件、電話、釘釘告警通知。

DataX支持一站式解決方案，支持常見數據源整庫全增量到MaxCompute、Hologres、ElasticSearch、DataHub等，同時能夠滿足分庫分表，單表、整庫多表、DDL消息等復雜場景。

07阿里云數據集成解決方案系統

1. 離線數倉-整庫遷移方案

下面將詳細介紹一下阿里云數據集成解決方案系統，首先是離線數倉的整庫遷移解決方案，我們將數據集成中的一些典型場景，抽象為數據產品解決方案，可以幫助提升用戶效率，降低用戶使用成本。上圖展示出源頭數據庫中所有的表列表，直接選中需要的表，選擇對應的同步方式，比如每日增量或者每日全量，選擇分批上傳或者整批上傳的同步并發配置，就可以上傳到MaxCompute中，這種可視化操作可以滿足大多離線數據遷移場景。

2. 實時數倉-全增量解決方案

實時數倉的全增量解決方案，可以非常方便的將現有數據庫通過簡單的配置后，完成存量的全量遷移，以及后續增量的實時同步。支持在目標庫中建表、自動建立離線同步任務、自動建立實時任務、自動啟動離線任務、自動啟動實時任務、自動建立和啟動增量和全量的融合任務、全流程的監控和展示，支持子步驟異常重試。通過這種方案，可以讓用戶不用關注每個全量任務和實時任務的瑣碎配置細節。通過這一套解決方案，可以完成整個數據的全量、增量實時數據的同步。

08阿里云DataWorks和數據集成的關系

前面介紹了阿里云數據集成開源和商業架構和能力，接下來介紹一下阿里云DataWorks和數據集成的關系。DataWorks是阿里云提供的一站式開發、數據治理的平臺，融合了阿里云、阿里集團12年之久的數據中臺、數據治理的實踐經驗。數據集成是阿里云DataWorks核心的一部分，DataWorks向下支持各種不同的計算和存儲引擎，比如阿里大數據計算服務MaxCompute、開源大數據計算平臺E-MapReduce、實時計算Realtime Compute、圖計算引擎GraphCompute、交互式分析引擎MC-Hologres等，以及支持OSS、HDFS、DB等各種存儲引擎。這些不同的計算存儲引擎可以被阿里云DataWorks統一管理使用，后面可以基于這些引擎去做整個數據倉庫。

DataWorks內部劃分為7個模塊，最下面是數據集成，可以完成各種模式的數據同步。數據集成之上，是元數據中心，提供統一的元數據服務。任務調度中心可以執行任務調度服務，數據開發方面，不同的存儲引擎，比如實時計算和離線計算，其有著不同的開發模式，DataWorks支持離線開發和實時開發。同時DataWorks擁有一套綜合數據治理的解決方案，會有一個數據服務模塊，統一向上提供數據服務，對接各種數據應用。最后將DataWorks各種能力統一通過OpenAPI對外提供服務。

數據集成模塊是可以單獨對用戶提供服務，單獨使用的，并不需要了解和掌握所有DataWorks模塊就可以將數據同步作業配置和運行起來。

09智能實時數倉解決方案示例

下面介紹一個智能實時數倉解決方案實例，可以應用在電商、游戲、社交等大數據實時場景中。數據源有結構化數據和非結構化數據，非結構化數據可以通過DataHub數據總線做實時數據采集，之后借助數據集成來實時寫到Hologres中做交互式分析，也可以將數據實時寫入到MaxCompute中，進行歸檔和離線數據計算，另外Flink也可以消費訂閱數據，做實時數據計算。Flink計算結果同時又可以寫入Hologres中，也可以將實時計算結果做實時大屏和實時預警。結構化數據也可以通過實時數據抽取或者批量數據采集方式，統一采集到DataWorks，實時數據可以寫入到Hologres或者定期歸檔到MaxCompute，離線數據可以通過批量數據加工到MaxCompute中來，另外MaxCompute和Hologres可以結合使用，進行實時聯邦查詢。

上面這套解決方案可以將阿里云實時數倉全套鏈路與離線數據無縫銜接，滿足一套存儲、兩種計算（實時計算和離線計算）的高性價比組合。