大資料相關工具

Maxwell資料實時同步工具

Maxwell 簡介

• Maxwell 是一個能實時讀取 MySQL 二進制日志檔案binlog，并生成 Json格式的訊息，作為生產者發送給 Kafka、Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、檔案或其它平臺的應用程式，它的常見應用場景有ETL、維護快取、收集表級別的 DML 指標、增量到搜索引擎、資料磁區遷移、切庫binlog回滾方案等，
• 官網地址：http://maxwells-daemon.io
• GitHub地址：https://github.com/zendesk/maxwell
• Maxwell 主要功能：
  • 支持 select * from table 方式進行全量資料初始化；
  • 支持在主庫發生 failover 后，自動恢復 binlog 位置，實作斷點續傳；
  • 可以對資料進行磁區，解決資料傾斜問題，發送到 Kafka 的資料支持庫、表、列等級別的資料磁區；
  • 作業方式是偽裝為 slave 接收 binlog events，然后根據 schema 資訊拼裝，可以接受 ddl、xid、row 等event，

MySQL Binlog 介紹

1. Binlog簡介

• MySQL 中有以下幾種日志：
  • 錯誤日志：記錄在啟動，運行或停止mysqld時遇到的問題
  • 通用日志：記錄建立的客戶端連接和執行的陳述句
  • 二進制日志(bin log)：記錄更改資料的陳述句
  • 中繼日志：從服務器 復制 主服務器接收的資料更改
  • 慢查詢日志：記錄所有執行時間超過 long_query_time 秒的所有查詢或不使用索引的查詢
  • DDL 日志(元資料日志)：元資料操作由 DDL 陳述句執行
• 在默認情況下，系統僅僅打開錯誤日志，關閉了其他所有日志，以達到盡可能減少IO損耗提高系統性能的目的，但是在一般稍微重要一點的實際應用場景中，都至少需要打開二進制日志，因為這是MySQL很多存盤引擎進行增量備份的基礎，也是MySQL實作復制的基本條件，
• MySQL 的二進制日志 binlog 可以說是 MySQL 最重要的日志，它記錄了所有的 DDL 和 DML 陳述句（除了資料查詢陳述句select、show等），以事件形式記錄，還包含陳述句所執行的消耗的時間，MySQL的二進制日志是事務安全型的，binlog 的主要目的是復制和恢復，
  • MySQL主從復制：MySQL Replication 在 Master 端開啟 binlog，Master 把它的二進制日志傳遞給slaves 來達到 master-slave 資料一致的目的，
  • 資料恢復：通過使用 mysqlbinlog 工具來使恢復資料，

2. Binlog的日志格式

• 記錄在二進制日志中的事件的格式取決于二進制記錄格式，支持三種格式型別：
  • Statement：基于 SQL 陳述句的復制（statement-based replication, SBR）
  • Row：基于行的復制（row-based replication, RBR）
  • Mixed：混合模式復制（mixed-based replication, MBR）
• Statement：
  • 每一潭訓修改資料的sql都會記錄在binlog中，
  • 優點：不需要記錄每一行的變化，減少了binlog日志量，節約了IO，提高了性能，
  • 缺點：在進行資料同步的程序中有可能出現資料不一致，比如 update tt set create_date=now()，如果用binlog日志進行恢復，由于執行時間不同可能產生的資料就不同，
• Row：
  • 它不記錄sql陳述句背景關系相關資訊，僅保存哪條記錄被修改，
  • 優點：保持資料的絕對一致性，因為不管sql是什么，參考了什么函式，它只記錄執行后的效果，
  • 缺點：每行資料的修改都會記錄，最明顯的就是update陳述句，導致更新多少條資料就會產生多少事件，占用較大空間，
• Mixed：
  • 從5.1.8版本開始，MySQL提供了Mixed格式，實際上就是Statement與Row的結合，在Mixed模式下，一般的復制使用Statement模式保存binlog，對于Statement模式無法復制的操作使用Row模式保存binlog， MySQL會根據執行的SQL陳述句選擇日志保存方式（因為statement只有sql，沒有資料，無法獲取原始的變更日志，所以一般建議為Row模式)，
  • 優點：節省空間，同時兼顧了一定的一致性，
  • 缺點：還有些極個別情況依舊會造成不一致，另外statement和mixed對于需要對binlog的監控的情況都不方便，

Mysql 實時資料同步方案對比

• mysql 資料實時同步可以通過決議mysql的 binlog 的方式來實作，決議binlog可以有多種方式，可以通過canal，或者maxwell等各種方式實作，以下是各種抽取方式的對比介紹：

• 其中canal由 Java開發，分為服務端和客戶端，擁有眾多的衍生應用，性能穩定，功能強大；canal 需要自己撰寫客戶端來消費 canal 決議到的資料，
• Maxwell相對于canal的優勢是使用簡單，Maxwell比Canal更加輕量級，它直接將資料變更輸出為json字串，不需要再撰寫客戶端，對于缺乏基礎建設，短時間內需要快速迭代的專案和公司比較合適，
• 另外Maxwell 有一個亮點功能，就是Canal只能抓取最新資料，對已存在的歷史資料沒有辦法處理，而Maxwell有一個bootstrap功能，可以直接引匯出完整的歷史資料用于初始化，非常好用，

開啟MySQL的Binlog

• mysql的版本盡量不要太低，也不要太高，最好使用5.6及以上版本，
• 添加mysql普通用戶maxwell，因為maxwell這個軟體默認用戶使用的是maxwell這個用戶，

-- 校驗級別最低，只校驗密碼長度
mysql> set global validate_password_policy=LOW;
mysql> set global validate_password_length=6;

-- 創建maxwell庫（啟動時候會自動創建，不需手動創建）和用戶
mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%'; 

-- 重繪權限
mysql> flush privileges;

• 修改組態檔 sudo vim /etc/my.cnf，添加或修改以下三行配置：

# binlog日志名稱前綴
log-bin=/var/lib/mysql/mysql-bin
# binlog日志格式
binlog-format=ROW
# 唯一標識，這個值的區間是：1到(2^32)-1
server_id=1

• 重啟 MySQL 服務

sudo service mysqld restart

• 驗證 binlog 是否配置成功

show variables like '%log_bin%';

• 查看binlog日志檔案生成：進入 /var/lib/mysql 目錄，查看binlog日志檔案

Maxwell 安裝部署

• 安裝包下載&解壓：

wget https://github.com/zendesk/maxwell/releases/download/v1.21.1/maxwell-1.21.1.tar.gz
tar -zxvf maxwell-1.21.1.tar.gz -C /bigdata/install

• 修改maxwell組態檔：

cd /bigdata/install/maxwell-1.21.1/
cp config.properties.example config.properties
vim config.properties

• 組態檔config.properties 內容如下：

# choose where to produce data to
producer=kafka
# list of kafka brokers
kafka.bootstrap.servers=node01:9092,node02:9092,node03:9092
# mysql login info
host=node03
port=3306
user=maxwell
password=123456
# kafka topic to write to
kafka_topic=maxwell

Maxwell 實時采集案例

• 關于 kafka 的安裝和使用，請參考我之前的博文：JavaEE 企業級分布式高級架構師（十二）Kafka學習筆記
• 啟動kafka集群和zookeeper集群

zkServer.sh start
bin/kafka-server-start.sh -daemon config/server.properties

• 創建 topic

# /usr/apps/kafka
bin/kafka-topics.sh --create --bootstrap-server 192.168.254.128:9092 --replication-factor 2 --partitions 3 --topic maxwell
# 查看topic
bin/kafka-topics.sh --list --bootstrap-server 192.168.254.128:9092

• node01 啟動 kafka 自帶控制臺消費者

bin/kafka-console-consumer.sh --bootstrap-server 192.168.254.128:9092 --topic maxwell --from-beginning

• node03 啟動maxwell服務

maxwell

• 插入資料進行測驗，向mysql表中插入一條資料，查看kafka是否能夠接收到資料

CREATE DATABASE /*!32312 IF NOT EXISTS*/`test_db` /*!40100 DEFAULT CHARACTER SET utf8 */;

USE `test_db`;

/*Table structure for table `user` */

DROP TABLE IF EXISTS `user`;

CREATE TABLE `user` (
  `id` varchar(10) NOT NULL,
  `name` varchar(10) DEFAULT NULL,
  `age` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

/*Data for the table `user` */
-- 插入資料
insert  into `user`(`id`,`name`,`age`) values  ('1','xiaokai',20);
-- 修改資料
update `user` set age= 30 where id='1';
-- 洗掉資料
delete from `user` where id='1';

• 輸出 json 資料欄位說明：