准备环境

maxwell-1.25.0.tar.gz
mysql-5.7.16

修改配置文件

开启 MySQL binlog

开启 MySQL binlog

创建 MySQL 用户

  1. # 在数据库中建立一个maxwell库用于存储Maxwell的元数据
  2. CREATE DATABASE maxwell;
  4. # 分配一个账号可以操作该数据库
  5. set global validate_password_length=4;
  6. set global validate_password_policy=0;
  7. GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY 'maxwell';
  9. # 分配这个账号可以监控其他数据库的权限
  10. GRANT  SELECT ,REPLICATION SLAVE , REPLICATION CLIENT  ON *.* TO maxwell@'%';

修改 Maxwell 配置

  1. cp config.properties.example config.properties
  1. producer=kafka
  2. kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
  3. #需要添加
  4. kafka_topic=gmall-maxwell
  5. # mysql login info
  6. host=hadoop102
  7. user=maxwell
  8. password=maxwell
  9. #需要添加 后续初始化会用
  10. client_id=maxwell_1
  11. # 分区方式，默认database，库的数据全部发送到一个分区
  12. producer_partition_by=primary_key
  13. #[ database|table|primary_key|random| column]

启动、测试

启动 Maxwell

编写 maxwell.sh 启动脚本

  1. /opt/module/maxwell-1.25.0/bin/maxwell --config  /opt/module/maxwell-1.25.0/config.properties >/dev/null 2>&1 &

image.png
image.png

测试

启动Kafka消费客户端，观察结果

  1. bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic gmall-maxwell

image.png

历史数据同步

maxwell-bootstrap 初始化数据

  1. bin/maxwell-bootstrap --user maxwell  --password maxwell --host hadoop102  --database gmall --table base_province --client_id maxwell_1
—user 数据库分配的操作maxwell数据库的用户名
—password 数据库分配的操作maxwell数据库的密码
—host 数据库主机名
—database 数据库名
—table 表名
—client_id maxwell-bootstrap不具备将数据直接导入kafka或者hbase的能力，通过—client_id指定将数据交给哪个maxwell进程处理，在maxwell的conf.properties中配置 
  1. create table GMALL.province_info
  2. (id varchar primary key,info.name varchar,info.area_code varchar,info.iso_code varchar)
  3. SALT_BUCKETS = 3;
  1. create table GMALL.user_info
  2. (id varchar primary key,user_level varchar,birthday varchar,gender varchar,age_group varchar,gender_name varchar)
  3. SALT_BUCKETS = 3;