任務需求：將MySQL里的數據實時增量同步到Kafka
1、準備工作
1.1、MySQL方面：開啟BinLog
1.1.1、修改my.cnf文件
vi /etc/my.cnf
[mysqld]
server-id = 1
binlog_format = ROW
1.1.2、重啟MySQL,然后登陸到MySQL之后，查看是否已經修改過來:
mysql> show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
1.1.3、創建Maxwell用戶，并賦予 maxwell 庫的一些權限
CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';
1.2、Kafka準備工作
1.2.1、啟動Zookeeper
zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties
1.2.2、啟動kafka
kafka-server-start /usr/local/etc/kafka/server.properties
2、Maxwell
2.1、下載安裝包
https://github.com/zendesk/maxwell/releases/download/v1.20.0/maxwell-1.20.0.tar.gz
2.2、解壓到指定位置
tar -zxvf maxwell-1.20.0.tar.gz
2.3、在MYSQL中創建測試用表（前提你要進入一個庫）
CREATE TABLE `test` (`id` bigint(20) NOT NULL AUTO_INCREMENT,`age` int(11) DEFAULT NULL,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
2.4、打開Maxwell（要在maxwell安裝目錄下）
bin/maxwell --user=maxwell --password=123456 --host='127.0.0.1' --producer=stdout
2.5、對數據進行操作
insert into test values(1,22,"小明");
update test set name='whirly' where id=1;
delete from test where id=1;
可以看到Maxwell控制臺的輸出，測試成功！
{"database":"test","table":"test","type":"insert","ts":1552153502,"xid":832,"commit":true,"data":{"id":1,"age":22,"name":"小明鋒"}}
{"database":"test","table":"test","type":"update","ts":1552153502,"xid":833,"commit":true,"data":{"id":1,"age":22,"name":"whirly"},"old":{"name":"小明"}}
{"database":"test","table":"test","type":"delete","ts":1552153502,"xid":834,"commit":true,"data":{"id":1,"age":22,"name":"whirly"}}
3、實現MySQL數據實時增量同步到Kafka
3.1、開啟指定到Kafka的MaxWell
bin/maxwell --user='maxwell' --password='123456' --host='127.0.0.1' \--producer=kafka --kafka.bootstrap.servers=localhost:9092 --kafka_topic=maxwell --kafka_version=0.10.2.1
3.2、對數據庫進行操作
insert into test values(1,22,"小明");
update test set name='whirly' where id=1;
delete from test where id=1;
3.3、啟動一個消費者來消費 maxwell topic的消息，觀察其輸出；
kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic test --from-beginning
3.4、再次執行數據庫結果觀察，仍然可以得到相同的輸出
