實時資料同步服務(canal+kafka)的架構設計

1. 移山是禧云自研的資料遷移平臺，包含異構資料源的遷移、實時資料同步等服務，有興趣的可以看這里：了解在移山中怎么實作異構資料源的遷移；

2. 本文主要介紹移山實時資料同步服務產生的背景以及整體架構設計，

3. 可以訪問 這里 查看更多關于大資料平臺建設的原創文章，

一. 移山實時資料同步服務產生背景

• 禧云各個子公司業務系統基本都是以 MySQL 為主；

• 做為資料支持部門，需要訂閱這些業務資料做為資料倉庫的資料源，來進行下游的資料分析，比如：

  • 各種離線資料 T+1 報表展示；

  • 實時資料大屏展示等，

微信小程式實時資料指標展示

像這種常見的實時資料指標大屏展示，背后可能就用到實時資料同步服務技術堆疊，

二. 移山實時資料同步服務使用canal中間件

1. 使用場景符合

它可以對 MySQL 資料庫增量日志決議，提供增量資料訂閱和消費，完全符合我們的使用場景，

2. 支持將訂閱到的資料投遞到kafka

canal 1.1.1版本之后，server端可以通過簡單的配置就能將訂閱到的資料投遞到MQ中，目前支持的MQ有kafka、RocketMQ，替代老版本中必須通過手動編碼投遞的方式，

移山的實時資料同步服務使用的MQ為kafka，以下為主要配置：

修改canal.properties中配置

# 這里寫上當前canal server所在機器的ip
canal.ip = 10.200.*.109
# register ip to zookeeper（這里寫上當前canal server所在機器的ip）
canal.register.ip = 10.200.*.109
# 指定注冊的zk集群地址
canal.zkServers =10.200.*.109:2181,10.200.*.110:2181
 
# tcp, kafka, RocketMQ（設定serverMode模式，這個配置非常關鍵，我們設定為kafka）
canal.serverMode = kafka
 
# 這個demo就是conf目錄里的實體
canal.destinations = demo
# HA模式必須使用該xml，需要將相關資料寫入zookeeper,保證資料集群共享
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
 
# 這里設定 kafka集群地址(其它關于mq的配置引數可以根據實際情況設定)
canal.mq.servers = 10.200.*.108:9092,10.200.*.111:9092

修改demo.properties中配置

# canal偽裝的MySQL slave的編號，不能與MySQL資料庫和其他的slave重復
# canal.instance.MySQL.slaveId=1003
# 按需修改成自己的資料庫資訊
# position info（需要訂閱的MySQL資料庫地址）
canal.instance.master.address=10.200.*.109:3306
 
# 這里配置要訂閱的資料庫，資料庫的用戶名和密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.defaultDatabaseName =
 
# 設定要訂閱的topic名稱
canal.mq.topic=demo
 
# 設定訂閱散列模式的磁區數
canal.mq.partitionsNum=3

備注

• 更多關于mq的配置引數解釋，可以訪問這里：https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

• 多個 canal server 除了 ip 和 MySQL.slaveId 設定不同外，其它都應該保持相同的配置，

3.支持帶cluster模式的客戶端鏈接，保障服務高可用

• 客戶端可以直接指定zookeeper地址、instance name，canal client 會自動從zookeeper中的running節點，獲取當前canal server服務的作業節點，然后與其建立鏈接；

• 其它canal server節點則做為Standby狀態，如果當前active節點發生故障，可以自動完成failover切換，

對canal 的高可用（HA機制）想了解更多，可以查看這篇文章，

三. 移山實時資料同步流程圖

實時資料同步服務流程圖（摘自《禧云數芯大資料平臺技術白皮書》）如下：

總結

• canal server 訂閱業務系統的 MySQL 資料庫產生的 bin log；

• canal server 將訂閱到的 bin log 投遞至 kafka 指定的topic里；

• kafka 消費端拿到訊息，根據實際的資料使用場景，將資料再寫入 Hbase 或 MySQL，或直接做實時分析，

四. 創建一個實時資料同步任務的主要步驟

以創建一個資料訂閱型別為 HBase 的資料同步任務為例，主要步驟如下：

1. 創建kafka的topic；

2. 進入到canal server的bin目錄，拷貝example整個目錄，生成一個新的實體目錄；

3. 手動修改新實體的組態檔，配置以下主要引數：

   • 3.1 設定slaveId，不能與已經成功運行的實體設定的slaveId值重復；

   • 3.2 要訂閱的資料庫所在的機器地址和埠號；

   • 3.3 要訂閱的資料庫名稱；

   • 3.4 要訂閱的表；

   • 3.5 要訂閱的資料庫用戶名、密碼；

   • 3.6 配置向kafka發送訊息的topic；

   • 3.7 配置kafka的partition等；

4. 重啟canal server；

5. 查看實體的啟動日志，判斷實體是否啟動成功，

存在的問題

由于缺乏 WebUI 的支撐，因此會存在以下問題：

• 流程復雜：如上這些一系列的操作都是依靠腳本的方式配置完成，配置程序繁瑣，資料開發者很容易在某個環節上發生遺漏、出錯；

• 不利于任務的統一管理：比如誰開發的任務可能只有寫代碼的這個人比較熟悉；

• 不方便查看任務的運行情況：比如已消費訊息數、延遲訊息數；

• 不利于排查問題：查看任務的執行情況只能登陸 canal server 所在服務器去查看任務所屬實體的啟動日志，如果遇到錯誤時，不能夠快速及時的排查問題，

怎么解決問題

為了解決上面提到的這些問題，我們開發了移山的實時資料同步服務，

后話

• 在最新的穩定版：canal 1.1.4版本，迎來最重要的 WebUI 能力；

• instance 可以通過 WebUI 來創建，但是有部分使用者反饋，instance的啟動會有不穩定的情況出現，我們期待穩定版本可以快速發布，

五. 移山實時資料同步服務整體架構

1. 所需集群環境

zookeeper集群

為什么要用zookeeper集群，可以看這篇文章：阿里canal是怎么通過zookeeper實作HA機制的？

kafka集群

• kafka 具有高吞吐量、內置的磁區、備份冗余分布式等特點，為大規模訊息處理提供了一種很好的解決方案；

• 前面已經提到過 canal 中間件通過簡單的配置即可支持將訂閱到的資料直接投遞到 kafka中，

canal server集群

為保障資料訂閱服務的穩定性，我們需要借助 canal 的HA機制，實作故障自動轉移，保障服務高可用，因此我們需要部署多個 canal server，

hbase集群

• 資料湖在禧云的實踐是存盤集團各子公司、ISV各種各樣原始資料的大型倉庫，其中的資料可供存取、處理、分析和傳輸；

• 資料湖的技術解決方案，我們選擇的是 Apache HBase，

2. 移山實時資料同步架構設計

架構圖

canal server端

• 在canal server 的多個節點上手工創建、運行instance；

備注：

• 我們在配置instance相關引數時，不指定具體的資料庫，這樣該instance可以訂閱到該MySQL節點上的所有資料庫，然后在移山創建同步任務時再指定具體要訂閱的表，

移山

前端采用 Vue.js + Element UI，后端使用 SpringBoot 開發：

• 前端工程

  • 負責提供創建實時同步任務所需的 WebUI；

  • 提供豐富的任務運行監控功能，

• 后端工程

  • 負責前端工程的資料介面，會記錄目標表（實時同步任務訂閱到的資料最終存盤的目的地）的各種元資料資訊，

備注：

• 如果要創建的同步任務將資料存盤至MySQL，則需要提前人工干預創建MySQL表（MySQL資料庫由DBA統一管理）；

• 如果要創建的同步任務將資料存盤至Hbase，則在移山創建任務時，由后臺自動創建Hbase表，

變形金剛

處理訂閱資料存盤的java工程，分為三個可單獨部署的模塊：

• canal client服務

  • canal 客戶端，從 canal server 拿到資料，并將資料投遞至kafka，

• kafkaToHbase服務

  • kafka 消費端，負責將接收的訊息進行轉化處理，處理后的資料存盤至Hbase，

• kafkaToMySQL服務

  • kafka 消費端，負責將接收的訊息進行轉化處理，處理后的資料存盤至MySQL，

備注：

• 以上三個服務均支持命令列啟動、停止，

關注微信公眾號

歡迎大家關注我的微信公眾號閱讀更多文章：

標籤：架構設計

上一篇：實時資料同步服務(canal+kafka)是如何保證訊息的順序性？

下一篇：qt 實作MVC Api控制器開發 web api介面-連載【5】-企業級系統開發實戰連載系列 -技術堆疊(vue、element-ui、qt、c++、sqlite)