maxwell订阅MySQL的binlog日志并生产到Kafka传输数据

maxwell订阅binlog并生产到Kafka 1.zookeeper伪分布式安装 2.Kafka伪分布式安装

创建一个topic为maxwell

sh $KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper hadoop_zxy:2181 --replication-factor 2 --partitions 1 --topic maxwell

3.MySQL 3.1 开启binlog

> [root@hadoop_zxy software]# vim /etc/my.cnf
> 添加以下内容
log-bin = mysql-bin
binlog_format = row
server-id = 1
innodb_file_per_table = ON
skip_name_resolve = ON
> 注意：binlog_format一定要设置为row
> 修改好后一定要重启数据库
[root@hadoop_zxy scripts]# service mysqld restart

3.2 查看binlog状态

mysql> show variables like '%log_bin%';
+---------------------------------+--------------------------------+
| Variable_name                   | Value                          |
+---------------------------------+--------------------------------+
| log_bin                         | ON                             |
| log_bin_basename                | /var/lib/mysql/mysql-bin       |
| log_bin_index                   | /var/lib/mysql/mysql-bin.index |
| log_bin_trust_function_creators | OFF                            |
| log_bin_use_v1_row_events       | OFF                            |
| sql_log_bin                     | ON                             |
+---------------------------------+--------------------------------+
6 rows in set (0.00 sec)

mysql>

3.3 查看binlog日志

mysql> show master status;
+------------------+----------+--------------+------------------+-------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+-------------------+
| mysql-bin.000002 |   220628 |              |                  |                   |
+------------------+----------+--------------+------------------+-------------------+
1 row in set (0.00 sec)

> 注意：这里的File和Posititon在启动maxwell的使用会用到

3.4 创建maxwell用户并分配权限

CREATE USER 'maxwell'@'%' IDENTIFIED BY '@Zhou147258369';
create database maxwell DEFAULT CHARSET utf8 COLLATE utf8_general_ci;
GRANT ALL ON *.* TO 'maxwell'@'%';
GRANT ALL ON *.* TO 'maxwell'@'localhost' IDENTIFIED BY '@Zhou147258369' ;
GRANT ALL ON *.* TO 'maxwell'@'localhost.localdomain' IDENTIFIED BY '@Zhou147258369';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'localhost';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'localhost.localdomain';
GRANT USAGE ON *.* TO 'maxwell'@'localhost.localdomain' IDENTIFIED BY '@Zhou147258369' WITH GRANT OPTION;
grant all privileges on *.* to maxwell@localhost.localdomain identified by '@Zhou147258369';
flush privileges;

4.maxwell

maxwell安装包

[root@hadoop_zxy software]# tar -zxvf maxwell-1.22.1.tar.gz -C /zxy/apps/
[root@hadoop_zxy maxwell-1.22.1]# vim config.properties.example
>> 加入以下内容：
# tl;dr config
daemon=true
log_level=info

producer=kafka
kafka.bootstrap.servers=hadoop_zxy:9092
kafka_topic=maxwell

# mysql login info
host=ip地址
user=maxwell
password=@Zhou147258369
port=3306
schema_database=maxwell

# maxwell启动端口
http_port=9002
http_diagnostic=true
http_path_prefix=/
metrics_type=jmx|http|datadog

5.启动 5.1 启动zookeeper 5.2启动Kafka 5.3启动Kafka消费端 5.4启动maxwell

第一次启动 指定binlog

[root@hadoop_zxy maxwell-1.22.1]# bin/maxwell --conf config.properties.example --init_position=/var/lib/mysql/mysql-bin.000002:47565:0 &

binlog的路径可以通过find / -iname 'mysql-bin.000002'查找

47565为3.3查看binlog日志处显示

0是maxwell的监听

6.测试 6.1 MySQL插入数据

6.2 maxwell监听状态

6.3 Kafka消费者接收数据