目录

一：安装maxwell及配置

1.1 maxwell简介

1.2 maxwell输出数据格式

1.3 maxwell原理

1.3.1 MySQL二进制日志

 1.3.2 MySQL主从复制

 1.4 maxwell部署

1.5 配置MYSQL

1.5.1 启用MySQL Binlog

1.5.2 创建Maxwell所需数据库和用户

1.5.3 配置Maxwell

1.6 maxwell的使用

1.6.1 maxwell启停

二：采集信息通道

2.1 采集通道 maxwell配置

---

一：安装maxwell及配置

1.1 maxwell简介

Maxwell 是由美国Zendesk公司开源，用Java编写的MySQL变更数据抓取软件。它会实时监控Mysql数据库的数据变更操作（包括insert、update、delete），并将变更数据以 JSON 格式发送给 Kafka、Kinesi等流数据处理平台。

1.2 maxwell输出数据格式

注：Maxwell输出的json字段说明：

 

1.3 maxwell原理

Maxwell的工作原理是实时读取MySQL数据库的二进制日志（Binlog），从中获取变更数据，再将变更数据以JSON格式发送至Kafka等流处理平台。 

很简单，就是将自己伪装成slave，并遵循MySQL主从复制的协议，从master同步数据。

1.3.1 MySQL二进制日志

二进制日志（Binlog）是MySQL服务端非常重要的一种日志，它会保存MySQL数据库的所有数据变更记录。Binlog的主要作用包括主从复制和数据恢复。Maxwell的工作原理和主从复制密切相关。

 1.3.2 MySQL主从复制

MySQL的主从复制，就是用来建立一个和主数据库完全一样的数据库环境，这个数据库称为从数据库。

1）主从复制的应用场景如下：

（1）做数据库的热备：主数据库服务器故障后，可切换到从数据库继续工作。（备胎）

（2）读写分离：主数据库只负责业务数据的写入操作，而多个从数据库只负责业务数据的查询工作，在读多写少场景下，可以提高数据库工作效率。

2）主从复制的工作原理如下：

（1）Master主库将数据变更记录，写到二进制日志(binary log)中

（2）Slave从库向mysql master发送dump协议，将master主库的binary log events拷贝到它的中继日志(relay log)

（3）Slave从库读取并回放中继日志中的事件，将改变的数据同步到自己的数据库。

 Master主数据库–》写入二进制日志（binlog）–》Slave从数据库发送dump协议–》binlog events拷贝到中继日志（relay log）–》Slave读取中继日志，同步至自己的数据库。

 1.4 maxwell部署

注：Maxwell-1.30.0及以上版本不再支持JDK1.8。

注：此处使用教学版安装包，教学版对原版进行了改造，增加了自定义Maxwell输出数据中ts时间戳的参数，生产环境请使用原版。

（1）将安装包解压至/opt/module

[axing@hadoop102 maxwell]$ tar -zxvf maxwell-1.29.2.tar.gz -C /opt/module/

（2）修改文件夹名称

[axing@hadoop102 module]$ mv maxwell-1.29.2/ maxwell

1.5 配置MYSQL

1.5.1 启用MySQL Binlog

MySQL服务器的Binlog默认是未开启的，如需进行同步，需要先进行开启。

1）修改MySQL配置文件/etc/my.cnf

[axing@hadoop102 ~]$ sudo vim /etc/my.cnf

 2）增加如下配置

[mysqld]

#数据库id
server-id = 1
#启动binlog，该参数的值会作为binlog的文件名
log-bin=mysql-bin
#binlog类型，maxwell要求为row类型
binlog_format=row
#启用binlog的数据库，需根据实际情况作出修改
binlog-do-db=gmall

注：MySQL Binlog模式

Statement-based：基于语句，Binlog会记录所有写操作的SQL语句，包括insert、update、delete等。

优点： 节省空间

缺点： 有可能造成数据不一致，例如insert语句中包含now()函数。

Row-based：基于行，Binlog会记录每次写操作后被操作行记录的变化。

优点：保持数据的绝对一致性。

缺点：占用较大空间。

mixed：混合模式，默认是Statement-based，如果SQL语句可能导致数据不一致，就自动切换到Row-based。

注：Maxwell要求Binlog采用Row-based模式。

 3）重启MySQL服务

[axing@hadoop102 ~]$ sudo systemctl restart mysqld

1.5.2 创建Maxwell所需数据库和用户

Maxwell需要在MySQL中存储其运行过程中的所需的一些数据，包括binlog同步的断点位置（Maxwell支持断点续传）等等，故需要在MySQL为Maxwell创建数据库及用户。

1）创建数据库

msyql> CREATE DATABASE maxwell;

2）调整MySQL数据库密码级别

mysql> set global validate_password_policy=0;
mysql> set global validate_password_length=4;

3）创建Maxwell用户并赋予其必要权限

//创建用户及密码
mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
//授予权限
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
//任何外部可以访问maxwell
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

1.5.3 配置Maxwell

1）修改Maxwell配置文件名称

[axing@hadoop102 maxwell]$ cd /opt/module/maxwell
[axing@hadoop102 maxwell]$ cp config.properties.example config.properties

2）修改Maxwell配置文件

[axing@hadoop102 maxwell]$ vim config.properties

#Maxwell数据发送目的地，可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
#目标Kafka集群地址
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092
#目标Kafka topic，可静态配置，例如:maxwell，也可动态配置，例如：%{database}_%{table}
kafka_topic=maxwell

#MySQL相关配置
host=hadoop102
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai

1.6 maxwell的使用

1.6.1 maxwell启停

1）启动Maxwell

[axing@hadoop102 ~]$ /opt/module/maxwell/bin/maxwell --config /opt/module/maxwell/config.properties --daemon

2）停止Maxwell

[axing@hadoop102 ~]$ ps -ef | grep maxwell | grep -v grep | grep maxwell | awk '{print $2}' | xargs kill -9

3）Maxwell启停脚本

#!/bin/bash

MAXWELL_HOME=/opt/module/maxwell

status_maxwell(){
    result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`
    return $result
}


start_maxwell(){
    status_maxwell
    if [[ $? -lt 1 ]]; then
        echo "启动Maxwell"
        $MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemon
    else
        echo "Maxwell正在运行"
    fi
}


stop_maxwell(){
    status_maxwell
    if [[ $? -gt 0 ]]; then
        echo "停止Maxwell"
        ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
    else
        echo "Maxwell未在运行"
    fi
}


case $1 in
    start )
        start_maxwell
    ;;
    stop )
        stop_maxwell
    ;;
    restart )
       stop_maxwell
       start_maxwell
    ;;
esac

二：采集信息通道

前提准备：打开zookeeper，kafka，maxwell

建议打开顺序：

1.zookeeper

2.kafka

3.maxwell

不然，kafka进程可能能会死掉，我的虚拟机是这样的。

2.1 采集通道 maxwell配置

1）修改Maxwell配置文件config.properties

[axing@hadoop102 maxwell]$ vim /opt/module/maxwell/config.properties

2）配置参数如下

log_level=info

producer=kafka
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092

#kafka topic配置
kafka_topic=topic_db

# mysql login info
host=hadoop102
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai

注意：这里配置的kafka_topic是topic_db，下文kafka消费数据时，接收的topic也应该是topic_db，前后一致。

3）重新启动Maxwell

[axing@hadoop102 bin]$ mxw.sh restart

 4）通道测试

（1）启动Zookeeper以及Kafka集群（如果已经打开就跳过这步）

（2）启动一个Kafka Console Consumer，消费topic_db数据

[axing@hadoop103 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_db

（3）生成模拟数据

[axing@hadoop102 bin]$ cd /opt/module/db_log/
[axing@hadoop102 db_log]$ java -jar gmall2020-mock-db-2021-11-14.jar 

这里的jar是一个脚本，运行jar包后会自动模拟生成数据。

（4）观察Kafka消费者是否能消费到数据