資料匯流排的資料寫入消費-SDK-資料匯流排 DataHub-阿里雲

介紹

DataHub 的 SDK 目前主要分為兩種，分別是 Low-Level 和 High-Level，Low-Level 實現了 DataHub 服務端定義的 API，一般適合做資源的管理；High-Level SDK 一般稱為 client-library，分為兩個大模組 Producer 和 Consumer，是對 Low-Level SDK 讀寫相關 API 的上層封裝，更合適做資料的讀寫。

Low-Level SDK 介紹

DataHub 是 RESTful 風格的 API ，Low-Level SDK 的每個介面都會對應一個服務端的 API。

以建立 Project 的 API 為例，API 如下所示。

  POST /projects/<ProjectName> HTTP/1.1
  x-datahub-client-version: 1.1
  Date: Tue, 08 May 2018 09:47:48 GMT
  Authorization: AuthorizationString
  Content-Type: application/json
  Content-Length: xxx
  {
      "Comment": "test project"
  }

對應的 JAVA SDK 的 API 為。

/**
* Create a datahub project.
*
* @param projectName The name of the project.
* @param comment     The comment of the project.
* @throws DatahubClientException Throws DatahubClientException
*/
CreateProjectResult createProject(String projectName, String comment);

High-Level SDK 介紹

Low-Level SDK 除了資源管理之外，最主要的功能就是寫入和消費，但是單個 API 是無法完成寫入和消費，需要多個 API 進行組合才能完成，並且每個 API 使用不合理都可能會造成導致最終無法讀寫，即使最終可以運行，但是使用不合理，也會給用戶端或者服務端造成一些不必要的壓力，出於這些考慮，我們在 Low-Level SDK 的基礎上，封裝了一層 High-Level SDK，最主要目的就是在讀寫這個最主要的情境下，簡化使用者操作，提升 SDK 易用性和穩定性。

Producer

寫入資料的情境，我們對比一下一下使用 Low-Level SDK 和 High-Level SDK 的區別，這裡只列舉使用到的 API。

使用 Low-Level SDK，發送資料步驟如下。

初始化 client
GetTopic 或者 ListSchema 擷取 schema 資訊
ListShard 擷取 shard 資訊，並篩選出 ACTIVE 狀態的 shard
資料發送前攢批
選擇一個 shard 調用 PutRecordsByShard 寫入資料
發送成功，繼續步驟 4
發送失敗，根據相應錯誤做出判斷，若可重試，重新步驟 5，若不可重試，做出相應處理
周期性調用 ListShard 更新 shard 資訊

使用 Producer ，發送資料步驟如下（以非同步方式發送為例）

初始化 Producer 並註冊回呼函數
非同步發送 sendAsync

從上面的對比可以看出，Producer 在寫入情境，更加易用和穩定，下面是 High-Level 相較於 Low-Level 的優勢整理。

邏輯簡單，易用性更強
穩定性更好，對於網路抖動等可重試異常，用戶端內部直接重試，並且重試次數可配置
自動感知擴縮容

Consumer

對於消費的情境，使用 Low-Level SDK 使用會更複雜，因為除了相關的 meta 資訊外，還需要解決幾個更棘手的問題。

每個 shard 都會有消費點位，點位需要提交到周期性提交到服務端，並在啟動的時候從服務端擷取
單個 shard 是無法在多個節點消費的，如果多個節點是的消費情境，那麼 shard 如何分配是一個非常棘手的問題
對於消費順序有嚴格要求的情境，如果發生 shard 分裂/合并，假設 shard0 分裂成了 shard1 和 shard2，這個時候 shard1 和 shard2 會平均分配 shard0 的 hash range，shard0 變為唯讀狀態，消費必須先消費完 shard0 的資料以後，再消費 shard1 和 shard2 才能嚴格保證順序

Low-Level SDK 消費流程：

初始化 client
GetTopic 或者 ListSchema 擷取 schema 資訊
ListShard 擷取 shard 資訊，並給每個任務手動分配需要消費的 shard
OpenOffsetSession 開啟消費，並擷取服務端儲存的 offset
通過 offset 調用 GetCusor 擷取消費的 cusor 資訊
根據 cursor 調用 GetRecords 讀取資料
讀取資料為空白，sleep 一段時間，繼續步驟 6
讀取資料不為空白，處理資料並更新 cursor，繼續步驟 6
周期性調用 ListShard 更新 shard 資訊，周期性調用 CommitOffset 更新消費點位

Consumer 消費流程

初始化 Consumer，初始化好以後會自動給每個節點分配好消費的 shard，並自動擷取服務端儲存的點位
調用 Read 讀取資料
讀取資料為空白，繼續步驟 2，用戶端內部控制間隔
讀取資料不為空白，處理資料，並繼續步驟 2

Consumer 的優勢除了易用性之外，最重要的是解決了上面提到的幾個比較棘手的問題。相對於 Low-Level SDK，Consumer 的優勢整理如下。

邏輯簡單，易用性更強
穩定性更好，對於網路抖動等可重試異常，用戶端內部直接重試，並且重試次數可配置
自動感知擴縮容，並可以完善處理好消費順序的問題
動態分配 shard，節點的加入和退出都可以動態調整 shard 分配，使用者無需做額外操作
自動維護點位資訊，使用者無需做額外操作。

SDK 狀態

語言	Low-Level	Hig-Level
JAVA	支援	支援
C++	支援	支援
Python	支援	支援
Golang	支援	暫不支援