基於Kafka Connect框架DataPipeline在實時數據集成上作了哪些提高？

在不斷知足當前企業客戶數據集成需求的同時，DataPipeline也基於Kafka Connect 框架作了不少很是重要的提高。

1. 系統架構層面。

DataPipeline引入DataPipeline Manager的概念，主要用於優化Source和Sink的全局化生命週期管理。當任務出現異常時，能夠實現對目的端和全局生命週期的管理。例如，處理源端到目的端讀取速率不匹配以及暫停等狀態的協同。

 

爲了增強系統的健壯性，咱們把Connector任務的參數保存在ZooKeeper中，方便任務重啓後讀取配置信息。

DataPipeline Connector經過JMX Client將統計信息上報Dashboard。在Connector中在技術上進行一些封裝，把一些通用信息，好比說Connector歷史讀取信息，跟管理相關的信息都採集到Dashboard裏面，提供給客戶。

2. 任務並行模式。

DataPipeline在任務並行方面作了一些增強。咱們在具體服務客戶的時候也遇到這樣的問題，須要同步數十張表。在DataPipeline Connector中，咱們容許每一個Task內部能夠定義和維護一個線程池，經過控制線程併發數，而且每一個Task容許設置行級別的IO控制。而對於JDBC類型的Task，咱們額外容許配置鏈接池的大小，減小上游和下游資源的開銷。

3. 規則引擎。

DataPipeline在基於Kafka Connect作應用時的基本定位是數據集成。數據集成過程當中，不該當對數據進行大量的計算，可是又不可避免地要對一些字段進行過濾，因此在產品中咱們也在考慮怎樣提供一種融合性。

雖然Kafka Connect提供了一個Transformation接口能夠與Source Connector和Sink Connector進行協同，對數據進行基本的轉換。但這是以Connector爲基本單位的，企業客戶須要編譯後部署到全部集羣的節點，而且缺少良好的可視化動態編譯調試環境支持。

 

基於這種狀況，DataPipeline產品提供了兩種可視化配置環境：基本編碼引擎（Basic Code Engine）和高級編碼引擎（Advanced Code Engine）。前者提供包括字段過濾、字段替換和字段忽略等功能，後者基於Groovy能夠更加靈活地對數據處理、而且校驗處理結果的Schema一致性。對於高級編碼引擎，DataPipeline還提供了數據採樣和動態調試能力。

 

4. 錯誤隊列機制。

咱們在服務企業客戶的過程當中也看到，用戶源端的數據永遠不會很「乾淨」。不「乾淨」的數據可能來自幾個方面，好比當文件類型數據源中的「髒記錄」、規則引擎處理特定數據產生未預期的異常、由於目的端Schema不匹配致使某些值沒法寫入等各類緣由。

面對這些狀況，企業客戶要麼把任務停下來，要麼把數據暫存到某處後續再處理。而DataPipeline採起的是第二種方式，經過產品中錯誤隊列預警功能指定面對錯誤隊列的策略，支持預警和中斷策略的設置和實施等，好比錯誤隊列達到某個百分比的時候任務會暫停，這樣的設置能夠保證任務不會因少許異常數據而中斷，被完整記錄下來的異常數據能夠被管理員很是方便地進行追蹤、排查和處理。企業客戶認爲，相比之前經過日誌來篩查異常數據，這種錯誤隊列可視化設置功能大大提高管理員的工做效率。

在作數據集成的過程當中，確實不該該對原始數據自己作過多的變換和計算。傳統ETL方案把數據進行大量的變換以後，雖然會產生比較高效的輸出結果，可是當用戶業務需求發生變化時，還須要從新創建一個數據管道再進行一次原始數據的傳輸。這種作法並不適應當前大數據分析的需求。

基於這種考慮，DataPipeline會建議客戶先作少許的清洗，儘可能保持數據的原貌。可是，這並非說，咱們不重視數據質量。將來的重要工做之一，DataPipeline將基於Kafka Streaming將流式計算用於數據質量管理，它不對數據最終輸出的結果負責，而是從業務角度去分析數據在交換過程當中是否發生了改變，經過滑動窗口去判斷到底數據發生了什麼問題，判斷條件是是否超出必定比例歷史均值的記錄數，一旦達到這個條件將進一步觸發告警並暫停同步任務。

總結一下，DataPipeline通過不斷地努力，很好地解決了企業數據集成過程須要解決異構性、動態性、可伸縮性和容錯性等方面的問題；基於Kafka Connect的良好基礎支撐構建了成熟的企業級數據集成平臺；基於Kafka Connect進行二次封裝和擴展，優化了應用Kafka Connect時面臨的挑戰：包括Schema映射和演進，任務並行策略和全局化管理等。將來，Datapipeline將會基於流式計算進一步增強數據質量管理。

更多關於實時數據集成和Kafka Connect的問題，歡迎直接訪問官方網址申請試用：www.datapipeline.com