​

導讀： 作為現代企業和組織機構的重要資源，資訊是進行科學管理與決策分析的基礎。ETL 則是把資料轉換為資訊、知識的關鍵步驟之一。在 AI 應用場景下，資料整合有哪些特點？隨著 AI 應用場景越來越多，為什麼我們需要從 ETL 轉換到 ELT？又會遇到哪些問題和挑戰？

本週三，陳肅博士與我們分享了《從 ETL 到 ELT——AI 時代資料整合的問題與解決方案》，AI 前線將分享內容整理成文，方便大家回顧參考。

大家好！很高興今天有機會和大家分享一些資料整合方面的看法和應用經驗。先自我介紹一下。我叫陳肅，博士畢業於中國科學院大學，資料探勘研究方向。現在北京數見科技（DataPipeline）任 CTO。之前曾經在中國移動研究院任職演算法工程師和使用者行為實驗室技術經理，之後作為合夥人加入過一家網際網路教育公司，從事智慧學習方面的研發工作。

​

在畢業後工作的這多年以來，我大部分時候在做大資料和機器學習相關的應用系統研發工作，資料的整合是其中一個非常重要的環節。加入 DataPipeline 後，公司研發的是一款企業級的資料整合產品，旨在幫助企業一站式解決資料整合和元資料管理問題。

ELT 和 ETL 是資料整合的兩種基本方式。前者專注於大資料的實時抽取和可靠傳輸，後者則包含了更豐富的資料轉換功能。 由於今天是和 AI 前線的朋友們一起探討資料整合，我主要結合 AI 應用的場景談談：為什麼 ELT 是更適合 AI 應用場景的資料整合方案、採用 Kafka 技術棧來構建 ELT 平臺所具備的優勢和問題以及我們所做的一些優化工作。希望能夠對大家的工作和學習有所幫助。

​

今天我的分享主要內容如上圖：

首先，我會介紹一下 AI 應用中資料整合的典型場景，ETL 和 ELT 兩種資料整合模式的異同點，以及為什麼 AI 應用下更適合採用 ELT 模式。然後，我會花一些篇幅介紹資料整合中需要重點考慮的基本問題，以及我們所採用的底層平臺——Kafka Connect 在解決這些問題上的優勢和侷限。

接下來，我會介紹 DataPipeline 對於 Kafka Connect 一些優化。有的是從底層做的優化，例如執行緒池的優化。有的則是從產品特性上的優化，例如錯誤資料佇列。

最後，我們談一談 Kafka Connect 和 Kafka Stream 的結合，以及我們用 Kafka Stream 做資料質量預警方面的一個應用 Case。

一、AI 應用場景下的資料整合

​

資料整合是把不同來源、格式、特點性質的資料在邏輯上或物理上有機地集中，為企業提供全面的資料共享。AI 是典型的資料驅動應用，資料整合在其中起著關鍵的基礎性作用。

以一個大家所熟悉的線上推薦服務為例，通常需要依賴三類資料：使用者的屬性 (年齡、性別、地域、註冊時間等）、商品的屬性（分類、價格、描述等）、使用者產生的各類行為（登入、點選、搜尋、加購物車、購買、評論、點贊、收藏、加好友、發私信等）事件資料。

隨著微服務框架的流行，這三類資料通常會存在於不同的微服務中：“使用者管理服務”儲存著使用者的屬性、好友關係、登入等資料；“商品管理服務”儲存的商品資訊；“訂單服務”儲存著使用者的訂單資料；“支付服務”儲存使用者的支付資料；“評論服務”記錄著使用者的評論和點贊資料。為了實現一個推薦服務，我們首先需要讓服務能夠訪問到這些資料。這種資料訪問應該是非侵入式的，也就是說不能對原有系統的效能、穩定性、安全性造成額外的負擔。因此，推薦服務不應當直接訪問這些分散的資料來源，而是應該通過某種方式將這些資料從各個業務子系統中提取出來，彙集到一個邏輯上集中的資料庫 / 倉庫，然後才能方便地使用機器學習框架（例如 Spark MLlib）來讀取資料、訓練和更新模型。

ETL 和 ELT 的區別與聯絡

​

資料整合包含三個基本的環節：Extract（抽取）、Transform（轉換）、Load（載入）。

抽取是將資料從已有的資料來源中提取出來，例如通過 JDBC/Binlog 方式獲取 MySQL 資料庫的增量資料；轉換是對原始資料進行處理，例如將使用者屬性中的手機號替換為匿名的唯一 ID、計算每個使用者對商品的平均打分、計算每個商品的購買數量、將 B 表的資料填充到 A 表中形成新的寬表等；載入是將資料寫入目的地。

根據轉換轉換髮生的順序和位置，資料整合可以分為 ETL 和 ELT 兩種模式。ETL 在資料來源抽取後首先進行轉換，然後將轉換的結果寫入目的地。ELT 則是在抽取後將結果先寫入目的地，然後由下游應用利用資料庫的聚合分析能力或者外部計算框架，例如 Spark 來完成轉換的步驟。

為什麼 ELT 更適合 AI 應用場景 

​

為什麼說 ELT 更適合 AI 的應用場景呢？

首先這是由 AI 應用對資料轉換的高度靈活性需求決定的。 絕大多數 AI 應用使用的演算法模型都包括一個特徵提取和變換的過程。根據演算法的不同，這個特徵提取可能是特徵矩陣的簡單的歸一化或平滑處理，也可以是用 Aggregation 函式或 One-Hot 編碼進行維度特徵的擴充，甚至特徵提取本身也需要用到其它模型的輸出結果。這使得 AI 模型很難直接利用 ETL 工具內建的轉換功能，來完成特徵提取步驟。此外，企業現在很少會從零構建 AI 應用。當應用包括 Spark/Flink MLlib 在內的機器學習框架時，內建的模型庫本身往往包含了特徵提取和變換的邏輯，這使得在資料提取階段就做複雜變換的必要性進一步降低；

其次，企業經常會基於同樣的資料構建不同應用。 以我之前所在的一家線上教育公司為例，我們構建了兩個 AI 的應用：其中一個是針對各類課程的推薦應用，主要用於增加使用者的購買轉化率。另外一個是自適應學習系統，用於評估使用者的知識掌握程度和題目的難度和區分度，從而為使用者動態地規劃學習路徑。兩個應用都需要使用者屬性、做題記錄、點選行為以及學習資料文字，但採用的具體模型的特徵提取和處理方式完全不同。如果用 ETL 模式，我們需要從源端抽取兩遍資料。而採用 ELT 模式，所有資料儲存在 HBase 中，不同的應用根據模型需要過濾提取出所需的資料子集，在 Spark 叢集完成相應的特徵提取和模型計算，降低了對源端的依賴和訪問頻次；

最後，主流的機器學習框架，例如 Spark MLlib 和 Flink MLlib，對於分散式、並行化和容錯都有良好的支援，並且易於進行節點擴容。 採用 ELT 模式，我們可以避免構建一個專有資料轉換叢集（可能還伴隨著昂貴的 ETL 產品 License 費用），而是用一個通用的、易於建立和維護的分散式計算叢集來完成所有的工作，有利於降低總體擁有成本，同時提升系統的可維護性和擴充套件性。

二、從 ETL 和 ELT 面臨的主要問題

​

採用 ELT 模式，意味著可以較少的關注資料整合過程中的複雜轉換，而將重點放在讓資料儘快地傳輸上。然而，一些共性的問題依然需要得到解決:

1. 資料來源的異構性： 傳統 ETL 方案中，企業要通過 ETL 工具或者編寫指令碼的方式來完成資料來源到目的地同步工作。當資料來源異構的時候，需要特別考慮 Schema（可以簡單理解為資料欄位型別）相容性帶來的影響。無論是 ETL 還是 ELT，都需要解決這一問題。

2. 資料來源的動態性： 動態性有兩方面含義。一是如何獲取資料來源的增量；二是如何應對資料來源端的 Schema 變化，例如增加列和刪除列。

3. 任務的可伸縮性： 當面對少量幾個資料來源，資料增量不過每日幾百 MB 的時候，ELT 平臺的可伸縮性不是什麼大問題。當 ELT 面對的是成百上千個數據源，或者資料來源資料增速很快時，ELT 平臺的任務水平切分和多工並行處理就成為一個必備的要求。平臺不僅要支援單節點的多工並行，還需要支援節點的水平擴充套件。此外，ELT 的上游通常會遇到一些吞吐能力較差的資料來源，需要能夠對讀取進行限速，避免對現有業務產生影響。

4. 任務的容錯性：ELT 平臺某些節點出現故障的時候，失敗的作業必須能夠遷移到健康的節點上繼續工作。同時，作業的恢復需要實現斷點重傳，至少不能出現丟失資料，最好能夠做到不產生重複的資料。

三、Kafka Connect 的架構

Kafka Connect：基於 Kafka 的 ELT 框架

​

可用於構建 ELT 的開源資料整合平臺方案不止一種，較廣泛採用的包括 Kafka Connect、DataX 等，也有公司直接採用 Flink 等流式計算框架。DataPipeline 作為一家提供企業資料整合產品的公司，我們在 Kafka Connect 之上踩了許多坑並且也做了許多優化。

四、踩過的坑與優化的點

 Kafka Connect 應用於 ELT 的關鍵問題 1

​

下面我們聊一聊 Kafka Connect 應用過程中的幾個關鍵問題。

首先是 任務的限速和資料快取問題。從 Kafka Connect 設計之初，就遵從從源端到目的地解耦性。當 Source 的寫入速度長時間大於 Sink 端的消費速度時，就會產生 Kafka 佇列中訊息的堆積。如果 Kafka 的 Topic Retention 引數設定不當，有可能會造成資料在消費前被回收，造成資料丟失。Kafka Connect 框架本身並沒有提供 Connector 級別的限速措施，需要進行二次開發。

 Kafka Connect 應用於 ELT 的關鍵問題 2

​

當用戶有多個數據源，或者單一資料來源中有大量的表需要進行並行同步時，任務的並行化問題 就產生了。Kafka Connect 的 rebalance 是牽一髮動全身，一個新任務的開始和停止都會導致所有任務的 reload。當任務數很多的時候，整個 Kafka Connect 叢集可能陷入長達數分鐘的 rebalance 過程。

解決的方法，一是用 CDC（Change Data Capture）來捕獲全域性的資料增量；二是 在任務內部引入多執行緒輪詢機制，減少任務數量並提高資源利用率。

 Kafka Connect 應用於 ELT 的關鍵問題 3

​

異構資料來源同步會遇到 Schema 不匹配 的問題。在需要精確同步的場景下（例如金融機構的異構資料庫同步），通常需要 Case by Case 的去定義對映規則。而在 AI 應用場景下，這個問題並不是很突出，模型訓練對於損失一點精度通常是可容忍的，一些資料庫獨有的型別也不常用。

 Kafka Connect 應用於 ELT 的關鍵問題 4

​

Source 端需要能夠檢測到 Schema 的變化，從而生成具有正確 Schema 格式的 Source Record。CDC 模式下，通過解析 DDL 語句可以獲取到。非 CDC 模式下，需要儲存一個快照才能夠獲取到這種變化。

下面我用一些時間對 DataPipeline 所做的優化和產品特性方面的工作。

DataPipeline 是一個底層使用 Kafka Connect 框架的 ELT 產品。首先，我們在底層上引入了 Manager 來進行全域性化的任務管理。Manager 負責管理 Source Connector 和 Sink Connector 的生命週期，與 Kafka Connect 的管理 API 通過 REST 進行互動。

系統的任何執行異常，都會進行統一的處理，並由通知中心傳送給任務的負責人和運維工程師。我們還提供了一個 Dashboard，用於圖形化方式對任務進行生命週期管理、檢索和狀態監控。使用者可以告別 Kafka Connect 的命令列。

DataPipeline 的任務並行模型

​

DataPipeline 在任務並行方面做了一些加強。在 DataPipeline Connector 中，我們在每個 Task 內部定義和維護一個執行緒池，從而能夠用較少的 Task 數量達到比較高的並行度，降低了 rebalance 的開銷。 而對於 JDBC 型別的 Connector，我們額外允許配置連線池的大小，減少上游和下游資源的開銷。此外，每個 Connector 還可以定義自己限速策略，以適應不同的應用環境需求。

DataPipeline 的錯誤佇列機制

​

通過產品中錯誤佇列預警功能，使用者可以指定面對錯誤資料暫存和處理邏輯，比如錯誤佇列達到某個百分比的時候任務會暫停，這樣的設定可以保證任務不會因少量異常資料而中斷，被完整記錄下來的異常資料可以被管理員非常方便地進行追蹤、排查和處理。

相比以前通過日誌來篩查異常資料，這種錯誤佇列視覺化功能能夠大大提升管理員的工作效率。

DataPipeline 的資料轉換

​

DataPipeline 實現了自己的 動態載入機制。提供了兩種 視覺化的轉換器：基本轉換器和高階轉換器。前者提供包括欄位過濾、欄位替換和欄位忽略等功能；後者基於 Java，可以更加靈活地對資料處理，並且校驗處理結果的 Schema 一致性。DataPipeline 還提供了資料取樣和動態除錯能力，方便使用者進行表級別的轉換規則開發。

​

值得注意的是，Kafka 不僅僅是一個訊息佇列系統，本身也提供了持久化能力。一個很自然的問題就是：能否不額外引入 Sink 端的外部儲存，直接從 Kafka 中獲取訓練資料？

如果模型本身要用到某個 Topic 的全量資料或者最近一段時間的資料，那麼通過設定合適的 retention 引數，可以直接將 Kafka 作為訓練資料的來源。Kafka 的順序讀模式可以提供非常高的讀取速度；如果模型要根據訊息的內容做資料篩選，那麼由於 Kafka 本身並不提供檢索能力，需要遍歷所有訊息，這樣就顯得比較低效了。

當模型用於線上時，可能還需要引入流式計算來完成實時特徵的提取工作。Kafka 本身就提供了這種流式計算能力。

流式計算在 ELT 中的作用 - 資料質量預警

​

DataPipeline 也將流式計算引入到平臺的質量預警功能中。在我們的未來版本中，使用者可以定義 Topic 級別的質量預警規則模型，例如“在 5 分鐘時間內，資料記錄的欄位 1 均值超過歷史均值記錄的比率超過 70%”為異常，採取策略為“告警並暫停同步”。通過這種方式，可以在 ELT 的過程中，儘早發現數據中的異常現象，避免大量異常資料進入資料目的地。

五、總結與展望

​

最後總結一下。資料整合並不是什麼新的概念，在過去二十多年間已經廣泛應用於各個行業的資訊系統。ELT 和 ETL 相比，最大的區別是“重抽取和載入，輕轉換”，從而可以用更簡單的技術棧、更輕量的方案搭建起一個滿足現代企業應用的資料整合平臺。AI 應用內在的特點也使得 ELT 特別適合這個場景。

Kafka Connect 本身是一個業界被廣泛採用的 ELT 框架，針對容錯、分散式、Schema 一致性等方面都提供了良好的支援，同時有大量的社群和商業資源可供參考和選擇。DataPipeline 基於 Kafka Connect 做了大量資料整合場景下的優化，與 Kafka Stream 相結合，能夠為包括 AI 在內的各種應用場景構建起一個完整的資料層支撐方案。

有其它關於資料整合的技術問題，也歡迎一起探討、共同提高。

參考資料

How to Build and Deploy Scalable Machine Learning in Production with Apache Kafka

https://www.confluent.io/blog/  

Kafka Connect 官方文件

https://docs.confluent.io/current/connect/index.html

Machine Learning + Kafka Streams Examples

https://github.com/kaiwaehner  

PredictionIO- 基於 Spark 的機器學習框架

http://predictionio.apache.org

Q & A

 Q1：DataPipeline 避開了資料處理這個過程，並以此提高效能，這個思路很認可。但是有個問題：從資料生產到資料利用的環節中，總要有一步資料處理的步驟的，這個步驟，從產品角度，DataPipeline 是如何考慮的？

A1：ELT 的核心思想就是要利用下游資料儲存效能大幅提升和機器學習應用的靈活性的優勢，在資料流轉的過程中不做過於複雜的計算。如果真的需要做處理，也可以基於我們的產品可以去寫轉換的程式碼。但這種處理都是無狀態的。有狀態處理，建議放到下游去做。這樣才更符合 ELT 的理念。

 Q2：請問資料的落地是自動的嗎？

A2： 基於原生 Kafka Connector，需要命令列啟動目標端型別的 Sink Connector，指定消費的 topic 列表，通過程式碼完成資料落地。基於 DataPipeline 產品，通過介面配置源和目的地後，落地是完全自動的。

 Q3：多執行緒讀，對源端的資料表或使用者許可權有沒有特定的要求？

A3：JDBC 模式的 Source Connector 使用的 RDBMS 使用者，需要具有選擇同步表的 select 許可權。CDC 模式的各不相同，參照產品內詳盡的許可權配置說明。

Q4：如何保證生產和消費的 EOS 剛好一次語義？

A4： Kafka Connect 下的 Exactly Once Semantic 依賴於具體 Connector 實現，Kafka Connect 框架本身對此只提供了必要非充分的支援。我們先來看 Source 端：假定 Source Connector 是從 MySQL 的 Binlog 中抽取資料到 Kafka，為了實現 EOS，首先 Source Connector 在每次提交記錄到 Kafka 的時候，需要原子化的記錄下來對應的 binlog position，這樣才能保證任務異常中斷、重啟後能夠從這個 position 繼續讀取。Kafka Connect 框架在 Source 端封裝了 offset storage 的儲存更新邏輯。offset storage 本質上是一個 Kafka 的 topic，利用 Kafka 的事務機制，理論上可以保證 offset 的修改和訊息傳送的原子性。再來看 Sink 端：如果 Sink Connector 可以將資料的輸出和 Offset 的記錄進行原子化操作，那麼同理也能夠做到 EOS。但這個原子化操作需要 Sink 端自己用某種機制實現，例如 Confluent 的 HDFS Connector 就用 WAL 日誌保證了寫入的 EO