当前位置: 首页 > news >正文

CentOS7环境Maxwell的安装及使用

目录

Maxwell的安装

下载安装包

解压安装包

配置环境变量

启用MySQL Binlog

创建Maxwell所需数据库和用户

配置Maxwell

Maxwell的使用

启动Kafka集群

Maxwell启停

Maxwell启停脚本

MySQL数据准备

Kafka开启消费者

全量数据同步

增量数据同步

启动Kafka消费者

修改数据

添加数据

删除数据

查询数据


Maxwell是用Java编写的MySQL变更数据抓取软件。 它会实时监控MySQL数据库的数据变更操作(insert、update、delete),并将变更数据以 JSON 格式发送给 Kafka、Kinesi等流数据处理平台。

Maxwell的安装

因为MySQL安装在node3机器(安装MySQL8),就近原则,所以Maxwell也安装在node3中。

下载安装包

官网下载安装包,下载版本如下:

maxwell-1.29.2.tar.gz

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

解压安装包

[hadoop@node3 installfile]$ tar -zxvf maxwell-1.29.2.tar.gz -C ~/soft

配置环境变量

[hadoop@node3 installfile]$ sudo vim /etc/profile.d/my_env.sh

添加如下内容

#MAXWELL_HOME
export MAXWELL_HOME=/home/hadoop/soft/maxwell-1.29.2
export PATH=$PATH:$MAXWELL_HOME/bin

让环境变量生效

source /etc/profile

启用MySQL Binlog

MySQL的Binlog默认是未开启的,如需捕获更新操作,需要先进行开启Binlog。

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

[hadoop@node3 installfile]$ sudo vim /etc/my.cnf

增加如下配置

server-id = 1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=gmall
binlog-do-db=test

配置项解释:

#数据库id
server-id = 1
#启动binlog,该参数的值会作为binlog的文件名
log-bin=mysql-bin
#binlog类型,maxwell要求为row类型
binlog_format=row
#启用binlog的数据库,需根据实际情况作出修改,如果需要开启多个数据库,直接再添加新的binlog-do-db设置行
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类型。

重启MySQL服务

[hadoop@node3 installfile]$ sudo systemctl restart mysqld

创建Maxwell所需数据库和用户

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

1)创建数据库

[hadoop@node3 installfile]$ mysql -uroot -p000000
...
省略若干输出
...
msyql> CREATE DATABASE maxwell;

2)创建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

1)基于模板配置文件复制得到Maxwell配置文件

[hadoop@node3 installfile]$ cd ~/soft/maxwell-1.29.2
​
[hadoop@node3 maxwell-1.29.2]$ ls
bin        config.properties.example                    lib      log4j2.xml     README.md
config.md  kinesis-producer-library.properties.example  LICENSE  quickstart.md
​
[hadoop@node3 maxwell-1.29.2]$ cp config.properties.example config.properties

2)修改Maxwell配置文件

[hadoop@node3 maxwell-1.29.2]$ vim config.properties

配置如下

#Maxwell数据发送目的地,可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
# 目标Kafka集群地址
kafka.bootstrap.servers=node2:9092,node3:9092,node4:9092
#目标Kafka topic,可静态配置,例如:maxwell,也可动态配置,例如:%{database}_%{table}
kafka_topic=topic_db# MySQL相关配置
host=node3
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

Maxwell的使用

启动Kafka集群

若Maxwell发送数据的目的地为Kafka集群,则需要先确保Kafka集群为启动状态。

开启Kafka集群 (Kafka集群安装)所有机器(node2、node3、node4),然后在任意一台机器执行如下命令:

[hadoop@node2 ~]$zk.sh start
[hadoop@node2 ~]$kf.sh start

Maxwell启停

1)启动Maxwell

maxwell --config $MAXWELL_HOME/config.properties --daemon

操作

[hadoop@node3 maxwell-1.29.2]$ maxwell --config $MAXWELL_HOME/config.properties --daemon
Redirecting STDOUT to /home/hadoop/soft/maxwell-1.29.2/bin/../logs/MaxwellDaemon.out
Using kafka version: 1.0.0
[hadoop@node3 maxwell-1.29.2]$ jps
1785 Maxwell
1839 Jps
[hadoop@node3 maxwell-1.29.2]$

2)停止Maxwell

ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9

操作

[hadoop@node3 maxwell-1.29.2]$ ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
[hadoop@node3 maxwell-1.29.2]$ jps
1891 Jps
[hadoop@node3 maxwell-1.29.2]$ 

Maxwell启停脚本

创建并编辑Maxwell启停脚本

[hadoop@node3 maxwell-1.29.2]$ cd ~/bin
[hadoop@node3 bin]$ vim mxw.sh

脚本内容如下

#!/bin/bashMAXWELL_HOME=/home/hadoop/soft/maxwell-1.29.2status_maxwell(){result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`return $result
}start_maxwell(){status_maxwellif [[ $? -lt 1 ]]; thenecho "启动Maxwell"$MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemonelseecho "Maxwell正在运行"fi
}stop_maxwell(){status_maxwellif [[ $? -gt 0 ]]; thenecho "停止Maxwell"ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9elseecho "Maxwell未在运行"fi
}case $1 instart )start_maxwell;;stop )stop_maxwell;;restart )stop_maxwellstart_maxwell;;
esac

注意:MAXWELL_HOME需要根据实际情况修改。

赋予权限

[hadoop@node3 bin]$ chmod 777 mxw.sh

启动Maxwell

[hadoop@node3 bin]$ mxw.sh start

查看进程

[hadoop@node3 bin]$ jps
1988 Jps
1942 Maxwell

停止Maxwell

[hadoop@node3 bin]$ mxw.sh stop

查看进程

[hadoop@node3 bin]$ jps
2015 Jps

MySQL数据准备

进入mysql命令行,执行如下命令

create database test;
use test;
create table stu (id int, name varchar(100), age int);
insert into stu values(1,"张三",18);
insert into stu values(1,"李四",20);

Kafka开启消费者

在node2、node3、node4集群任意一台执行如下消费者命令,这里在node2执行

kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topic_db

全量数据同步

Maxwell提供了bootstrap功能来进行历史数据的全量同步,命令如下:

[hadoop@node3 bin]$ mxw.sh start
​
[hadoop@node3 bin]$ maxwell-bootstrap --database test --table stu --config $MAXWELL_HOME/config.properties

kafka消费者输出如下

{"database":"test","table":"stu","type":"bootstrap-start","ts":1719415852,"data":{}}
{"database":"test","table":"stu","type":"bootstrap-insert","ts":1719415852,"data":{"id":1,"name":"张三","age":18}}
{"database":"test","table":"stu","type":"bootstrap-insert","ts":1719415852,"data":{"id":1,"name":"李四","age":20}}
{"database":"test","table":"stu","type":"bootstrap-complete","ts":1719415852,"data":{}}

格式化后如下:

{"database": "test","table": "stu","type": "bootstrap-start","ts": 1719415852,"data": {}
} {"database": "test","table": "stu","type": "bootstrap-insert","ts": 1719415852,"data": {"id": 1,"name": "张三","age": 18}
} {"database": "test","table": "stu","type": "bootstrap-insert","ts": 1719415852,"data": {"id": 1,"name": "李四","age": 20}
} {"database": "test","table": "stu","type": "bootstrap-complete","ts": 1719415852,"data": {}
}

(1)第一条type为bootstrap-start和最后一条type为bootstrap-complete的数据,是bootstrap开始和结束的标志,不包含数据,中间的type为bootstrap-insert的数据才包含数据。

(2)一次bootstrap输出的所有记录的ts都相同,为bootstrap开始的时间(系统时间)。

增量数据同步

启动Kafka消费者
kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topic_db

修改数据

将李四的id改为2

update stu set id=2 where name="李四";

kafka消费者输出

{"database":"test","table":"stu","type":"update","ts":1719416255,"xid":2595,"commit":true,"data":{"id":2,"name":"李四","age":20},"old":{"id":1}}

格式化输出

{"database": "test","table": "stu","type": "update","ts": 1719416255,"xid": 2595,"commit": true,"data": {"id": 2,"name": "李四","age": 20},"old": {"id": 1}
}

添加数据

例如:添加一条王五的数据

insert into stu values(3,"王五",23);

kafka消费者输出

{"database":"test","table":"stu","type":"insert","ts":1719416370,"xid":2853,"commit":true,"data":{"id":3,"name":"王五","age":23}}

格式化输出

{"database": "test","table": "stu","type": "insert","ts": 1719416370,"xid": 2853,"commit": true,"data": {"id": 3,"name": "王五","age": 23}
}

删除数据
delete from stu where id=3;

kafka消费者输出

{"database":"test","table":"stu","type":"delete","ts":1719416588,"xid":3339,"commit":true,"data":{"id":3,"name":"王五","age":23}}

格式化输出

{"database": "test","table": "stu","type": "delete","ts": 1719416588,"xid": 3339,"commit": true,"data": {"id": 3,"name": "王五","age": 23}
}

查询数据
select * from stu;

kafka消费者无新增的输出

可以看到,Maxwell可以监听到MySQL开启Binlog数据库的增、删、改操作。

完成!enjoy it!

http://www.lryc.cn/news/384605.html

相关文章:

  • python环境变量
  • 用进程和线程完成TCP进行通信操作及广播和组播的通信
  • 浅谈Tomcat
  • C++精解【7】
  • 堆箱子00
  • Linux 命令:iftop
  • web学习笔记(六十九)vue2
  • JavaScript全解:从基础到高级,掌握每一个知识点
  • RabbitMQ的Direct交换机
  • 2024.6.26 待学习知识点
  • 【LeetCode】每日一题:相交链表
  • 6.26.1 残差卷积变压器编码器的混合工作流程用于数字x线乳房x光片乳腺癌分类
  • [leetcode]avoid-flood-in-the-city 避免洪水泛滥
  • Pytorch基础
  • 嵌入技术Embedding
  • Pandas中的数据转换[细节]
  • vue2面试题——路由
  • 【AI应用探讨】—朴素贝叶斯应用场景
  • 使用matlab的大坑,复数向量转置!!!!!变量区“转置变量“功能(共轭转置)、矩阵转置(默认也是共轭转置)、点转置
  • 昇思25天学习打卡营第8天|保存与加载
  • 【vueUse库Animation模块各函数简介及使用方法】
  • 汇川H5u小型PLC作modbusRTU从站设置及测试
  • 基于Java的多元化智能选课系统-计算机毕业设计源码040909
  • idea使用maven打包报错GBK不可映射字符
  • 解决Linux系统Root不能远程SSH登录
  • 【java】【控制台】【javaSE】 初级java家教管理系统控制台命令行程序项目
  • (2024)豆瓣电影TOP250爬虫详细讲解和代码
  • am62x芯片安全类型确认(HS-SE, HS-FS or GP)
  • 高通安卓12-在源码中查找应用的方法
  • 民用无人驾驶航空器运营合格证怎么申请