实时数仓-Maxwell

一、Maxwell

maxwell是有美国zendesk开源，用Java编写的Mysql实时抓取软件；其抓取的原理也是基于binlog

二、Maxwell与Canal的对比

Maxwell没有Server+client模式，只有一个server把数据发送到消息队列或redis中；缺点：没有Canal灵活
Maxwell有一个亮点功能，Canal只能抓取到最新数据，对已存在的历史数据没有办法处理；而Maxwell有一个bootstrap功能，可以直接引导出完整的历史数据用于初始化，非常好用。
Maxwell不能支持HA，但是它支持断点还原，即错误解决后重启继续上次点儿读取数据。
Maxwell只支持json格式，而Canal如果用Server+client模式的话，可以自定义格式。
Maxwell比Canal更加轻量级

三、安装Maxwell

Maxwell官网地址和下载地址

官方网站：
http://maxwells-daemon.io/

Source：
https://github.com/zendesk/maxwell

安装包下载地址，我这边选择v1.25.0这一版本
https://github.com/zendesk/maxwell/releases/download/v1.25.0/maxwell-1.25.0.tar.gz

新建目录maxwell，并解压缩

新建目录
mkdir ./maxwell
解压缩
tar -zxvf maxwell-1.25.0.tar.gz

在mysql中新建maxwell用户名

CREATE USER 'maxwell'@'%' IDENTIFIED BY 'XXXXXX';

GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

修改配置文件，如下图，Maxwell只有一个配置文件，把config.properties.example复制一份，并修改名字为config.properties

image.png

```
复制文件
cp config.properties.example config.properties
```
**修改配置文件config.properties**
[图片上传失败...(image-bea128-1601021500215)]
```
把kafka.bootstrap.servers改成自己kafka的集群地址
host改成要监控的mysql数据库host
再加两条配置
1. 设置kafka的topic名称
kafka_topic=topicName
2. 设置Maxwell客户端id的唯一标识,这个对于使用Maxwell拉取历史数据进行初始化时有用
client_id=maxwell_1 
```

Maxwell需要建元数据库，maxwell

建数据库
CREATE DATABASE maxwell
分配一个账号，分配所有权限操作数据库maxwell
CRANT ALL ON maxwell.* to 'maxwell'@'%' IDENTIFIED BY 'maxwell';
访问别的数据库权限
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

启动Maxwell，启动时需要指定配置文件maxwell.properties位置!

image.png

```
后台启动Maxwell
./bin/maxwell --config ./maxwell.properties --kafka_partition_hash=default  --producer_partition_by=primary_key >/dev/null 2>&1 &

还可以开启多个服务，复制新的config.propertites文件，修改里面配置，topic和server_id, client_id设置不一样；如下语句直接把参数放在语句中
./maxwell --user='maxwell' --password='maxwell' --producer=kafka --host='dw-test-cluster-007' --port=3306 --kafka.bootstrap.servers=dw-test-cluster-004:9092 --kafka_topic=maxwell_test --client_id =maxwell_2 --daemon
下面语句是使用config.properties.v2进行启动
```
**注意server_id是根据mysql的配置文件/etc/my.cnf来的，如果要设置也应该与my.cnf中一致**
**使用jps查看是否启动成功，如下图所示**
[图片上传失败...(image-ac0f5a-1601021500215)]

启动后创建的maxwell元数据库中会自动新建表
[图片上传失败...(image-fed6d9-1601021500215)]

可以在监控数据库中写入数据，在kafka目录下查看对应的topic的值是否滚动出现
```
./kafka-console-consumer.sh --bootstrap-server dw-test-cluster-002:9092 --topic maxwell_binlog --from-beginning
```

在mysql中一条语句插入多条数据时，canal中在data对象中会用数组记录多条记录，Maxwell会在kafka中拆成多条记录，每条记录中data对象只会保存一条数据

保证写入kafka中数据分区

增加新的参数,依据什么进行分区，有[database, table, primary_key, random，transaction_id, column]这些选项
producer_partition_by=primary_key