mysql+canal+kafka實時計算

當用戶資料產生寫入到mysql, canal監控到mysql的寫操作，解析binlog日誌寫入到kafka。

canal簡介

canal [kə'næl]，譯意為水道/管道/溝渠，主要用途是基於 MySQL 資料庫增量日誌解析，提供增量資料訂閱和消費。

工作原理：
canal模擬MySQL slave的互動協議，偽裝自己為MySQL slave訂閱MySQL master的binlog，實現資料同步的中介軟體。

支援將訂閱到的資料投遞到kafka：
canal 1.1.1版本之後，server端可以通過簡單的配置就能將訂閱到的資料投遞到MQ中，目前支援的MQ有kafka、RocketMQ

canal部署

到 canal官網 下載服務端的安裝包：

將canal.deployer解壓到固定目錄：

tar zxvf canal.deployer-1.1.5.tar.gz  -C /opt/canal

配置修改引數

1. 修改instance 配置檔案canal/conf/example/instance.properties

# 需要同步的MySQL資料庫地址
canal.instance.master.address=127.0.0.1:3306

# 配置資料庫的使用者名稱和密碼
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

# 需要同步的資料庫名稱 
canal.instance.defaultDatabaseName = test 

# 設定要訂閱的topic名稱
canal.mq.topic= example
 

2. 修改canal 配置檔案canal/conf/canal.properties

# 當前canal server所在機器的ip
canal.ip = 127.0.0.1

# register ip to zookeeper
canal.register.ip = 127.0.0.1

# 可選項: tcp(預設), kafka, RocketMQ
canal.serverMode = kafka

# canal.properties配置檔案同級目錄下的example目錄
canal.destinations = example 

# 配置kafka對應的地址和埠
canal.mq.servers = 127.0.0.1:9092
 

啟動

cd ./canal/bin
sh ./start.sh