Maxwell单节点部署

1）主数据库的备用：主数据库服务器故障后，可切换到从数据库继续工作。
2）读写分离：主数据库只负责业务数据的写入操作，而多个从数据库只负责业务数据的查询工作，在读多写少场景下，可以提高数据库工作效率。

类似一个图书馆和一个分馆的关系。
1）主馆记录变化：主图书馆（Master）购入了新书，并将这些变更记录在一个清单上（相当于二进制日志）。
2）分馆请求更新：分馆（Slave）定期向主馆询问新书的清单，以便获取最新的书籍信息。这个过程中，分馆把收到的新书信息放到自己的临时清单上（相当于中继日志）。
3）应用变化：分馆根据这个临时清单将新书上架，以确保分馆的藏书与主馆的一样。

1）主库记录变化：当主数据库（Master）上发生数据变化（比如新增、更新或删除数据）时，这些变化会被记录到一个特殊的日志文件中，叫做二进制日志（binary log）。
2）从库请求数据：从数据库（Slave）会向主数据库请求这些日志数据，获取记录的数据变化。这些变化会被存储到从库的一个临时日志中，叫做中继日志（relay log）。
3）应用数据变化：接下来，从库会读取中继日志中的记录，并将这些变化应用到自己的数据库中，以确保从库的数据与主库保持一致。

cd /opt/module

tar -zxvf maxwell-1.29.2.tar.gz
mv maxwell-1.29.2/ maxwell

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

sudo vim /etc/my.cnf

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

Statement-based：
基于语句，Binlog会记录所有写操作的SQL语句，包括insert、update、delete等。
优点：节省空间
缺点：有可能造成数据不一致，例如insert语句中包含now()函数。

Row-based：
基于行，Binlog会记录每次写操作后被操作行记录的变化。
优点：保持数据的绝对一致性。
缺点：占用较大空间。

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

sudo systemctl restart mysqld
mysql -uroot -p000000

# 创建数据库
msyql> CREATE DATABASE maxwell;
# 创建Maxwell用户并赋予其必要权限
mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

# 修改Maxwell配置文件名称
cd /opt/module/maxwell
cp config.properties.example config.properties

# 修改Maxwell配置文件
vim config.properties

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

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

# 过滤gmall中的z_log表数据，该表是日志数据的备份，无须采集
filter=exclude:gmall.z_log
# 指定数据按照主键分组进入Kafka不同分区，避免数据倾斜
producer_partition_by=primary_key

vim mxw.sh

#!/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

mxw.sh start
mxw.sh stop