使用maxwell实时采集mysql数据

电玩女神 2022-11-08 14:25 257阅读 0赞

# 使用maxwell实时采集mysql数据 #

## 1. 什么是maxwell ##

maxwell 是由美国zendesk开源，用java编写的`Mysql实时抓取软件`。 其抓取的原理也是`基于binlog`。

## 2. Maxwell与canal的对比 ##

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

## 3. 使用maxwell前的准备工作 ##

1.  在mysql中创建一个数据库, 用于存储maxwell的元数据

CREATE DATABASE `maxwell` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';

1.  创建可以操作数据库maxwell的用户:maxwell

Mysql> set global validate_password_policy=0;
    	mysql> set global validate_password_length=4;
    
    	GRANT ALL ON maxwell.*  TO  'maxwell'@'%' IDENTIFIED BY 'aaaaaa';

1.  给用户maxwell分配操作其他数据库的权限

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

## 4. 安装和配置maxwell ##

1.  下载maxwell

wget https://github.com/zendesk/maxwell/releases/download/v1.27.1/maxwell-1.27.1.tar.gz

1.  解压

tar -zxvf maxwell-1.27.1.tar.gz -C /opt/module

1.  配置maxwell

cd /opt/module/maxwell-1.27.1
    	vim config.properties

添加如下配置:

# tl;dr config
    	log_level=info
    
    	producer=kafka
    	kafka.bootstrap.servers=hadoop162:9092,hadoop163:9092,hadoop164:9092
    	kafka_topic=ods_db
    	# 按照主键的hash进行分区, 如果不设置是按照数据库分区
    	producer_partition_by=primary_key
    
    	# mysql login info
    	host=hadoop162
    	user=maxwell
    	password=aaaaaa
    	# 排除掉不想监控的数据库
    	filter=exclude:gmall2021_realtime.*
    	# 初始化维度表数据的时候使用
    	client_id=maxwell_1

## 5. 启动maxwell ##

1.  启动maxwell

/opt/module/maxwell-1.27.1 » bin/maxwell --config config.properties --daemon

1.  确认kafka是否收到数据  
    起一个终端消费者:

bin/kafka-console-consumer.sh --bootstrap-server hadoop162:9092 --topic maxwell_gmall_db

在mysql中生成数据, 确认kafka是否收到数据.

## 6. Maxwell发送到kafka的数据格式 ##

{ 
        "database":"gmall",
        "table":"comment_info",
        "type":"insert",
        "ts":1598434438,
        "xid":27085,
        "commit":true,
        "data":{ 
            "id":1298554271214907454,
            "user_id":988,
            "sku_id":5,
            "spu_id":5,
            "order_id":548,
            "appraise":"1201",
            "comment_txt":"评论内容：78483837649887576216843442715245587379516398154672",
            "create_time":"2020-08-26 17:33:58",
            "operate_time":null
        }
    }

## 7. Maxwell的初始化数据功能 ##

对Mysql中的已有的旧数据, 如何导入到Kafka中? Canal无能为力, Maxwell提供了一个初始化功能, 可以满足我们的需求

bin/maxwell-bootstrap --user maxwell  --password aaaaaa --host hadoop162  --database gmall2021 --table 
    user_info --client_id maxwell_1

maxwell-bootstrap不具备将数据直接导入kafka或者hbase的能力，通过–client\_id指定将数据交给哪个maxwell进程处理，在maxwell的conf.properties中配置