DataPipeline數據融合重磅功能丨一對多實時分發、批

距離上個月的用戶體驗升級帖已經過去一個月了，在這一個月的時間里，DataPipeline的小伙伴都在忙些什么呢？

為能更好地服務用戶，DataPipeline最新版本支持：

1. 一個數據源數據同時分發（實時或定時）到多個目的地；
2. ?提升Hive的使用場景：

• 寫入Hive目的地時，支持選擇任意目標表字段作為分區字段；

  
  

• 可將Hive作為數據源定時分發到多個目的地。

3. 定時同步關系型數據庫數據時，可自定義讀取策略來滿足各個表的同步增量需求。

本篇將首先介紹一下一對多數據分發及批量讀取模式2.0的功能，后續功能會陸續發布。

推出背景

推出「一對多數據分發」的背景

在歷史版本中，DataPipeline每個任務只允許有一個數據源和目的地，從數據源讀取的數據只允許寫入到一張目標表。這會導致無法完美地支持客戶的兩個需求場景：

需求場景一：

客戶從一個API數據源或者從KafkaTopic獲取JSON數據后，通過高級清洗解析寫入到目的地多個表或者多個數據庫中，但歷史版本無法同時寫入到多個目的地，只能創建多個任務。這會導致數據源端會重復獲取同一批數據（而且無法完全保證數據一致性），浪費資源，并且無法統一管理。

需求場景二：

客戶希望創建一個數據任務，并從一個關系型數據庫表實時（或定時）分發到多個數據目的地。在歷史版本中，用戶需要創建多個任務來解決，但創建多個任務執行該需求時會重復讀取數據源同一張表的數據，比較浪費資源。客戶更希望只讀取一次便可直接解析為多個表，完成該需求場景。

新功能解決的問題：

1. 用戶在一個數據任務中選擇一個數據源后，允許選擇多個目的地或者多個表作為寫入對象，而不需要創建多個任務來實現該需求。

2. 用戶在單個任務中針對每個目的地的類型和特性，可以單獨設置各個目的地表結構和寫入策略，大大減少了數據源讀取次數和管理成本。

擴展Hive相關使用場景

歷史版本中，DataPipeline支持各類型數據源數據同步到Hive目的地的需求場景。但由于每個客戶的Hive使用方式、數據存儲方式不同，兩個需求場景在歷史版本中并沒有得到支持。

需求場景一：
動態分區字段。歷史版本中，用戶只允許選擇時間類型字段作為分區字段。在真實的客戶場景中除了按照時間做分區策略外，客戶希望指定Hive表任意字段作為分區字段。

需求場景二：
客戶希望除了以Hive作為目的地，定時寫入數據到Hive外，客戶還希望使用DataPipeline可以定時分發Hive表數據到各個應用系統，解決業務需求。

新功能解決的問題：

1. 允許用戶指定目的地Hive表中任何字段作為分區字段，并支持選擇多個分區字段。

2. 新增Hive數據源，可作為數據任務讀取對象。

推出「批量讀取模式2.0功能」的背景

需求場景：

關系型數據庫（以MySQL為例）的表沒有權限讀取BINLOG，但在業務上客戶需要定期同步增量數據，在權限只有SELECT情況下，需要做到增量數據的同步任務。

新版本出現之前，DataPipeline在用戶選擇批量讀取模式時提供了增量識別字段的功能，可以選擇自增字段或者更新時間字段作為條件完成增量數據的同步，但部分表可能沒有這種類型的字段，或者增量同步的邏輯不通（比如：只同步過去1小時的數據，或只同步到5分鐘前的數據等）。

新功能解決的問題：

?

1. 在關系型數據庫作為數據源的情況下，允許用戶針對每一個表設置WHERE讀取條件，并提供lastmax方法。

2. 使用該函數DataPipeline會取該任務下已同步數據中某一個字段的最大值，用戶可以使用該值作為WHERE語句讀取條件。

3. 用戶使用last_max（）函數，在首次執行該語句或對應字段暫無數值時，則會忽略該函數相關的讀取條件。

4. 允許用戶結合其他數據庫提供的方法編輯讀取條件：

例：以時間字段作為讀取條件，每次只同步一小時前的數據，且只同步未曾讀取的數據。
SELECT * FROMtable1?WHEREupdate_time?> 'last_max(update_time)' ANDupdate_time<= DATE_SUB(NOW(), INTERVAL 24 HOUR)?
相較于之前的一對一設置，新版本上線后用戶可以通過批量設置增量識別字段、批量移除、批量修改表名稱等，批量地操作一些表，批量地做一些動態修改，減少用戶配置成本。例如，現在需要在200張表的名稱后面都加一個data_warehouse，不同于以往的逐一添加，現在可以批量添加這些前綴后綴。?
DataPipeline每一次版本的迭代都凝聚了團隊對企業數據使用需求的深入思索，其它新功能還在路上，很快就會跟大家見面了，希望能夠切實幫助大家更敏捷高效地獲取數據。