文章实时采集(mysql实时数据采集框架maxwell基本介绍-苏州安嘉 )

　　文章实时采集(mysql实时数据采集框架maxwell基本介绍-苏州安嘉

)

　　介绍本文文章主要介绍MaxWell（Mysql实时数据采集）及相关经验和技巧。文章约13381字，浏览量317，点赞2，值得参考！

　　mysql实时数据采集框架maxwell基础介绍

　　maxwell 是一个可以实时读取 MySQL 二进制日志 binlog 并生成 JSON 格式消息的应用程序，这些消息作为生产者发送到 Kafka、Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其他平台。其常见的应用场景包括ETL、维护缓存、采集表级dml指标、递增到搜索引擎、数据分区迁移、切割数据库binlog回滚方案等。

　　Maxwell主要提供以下功能：

　　支持 SELECT * FROM 表进行 44 行全数据初始化。支持主库故障转移后binlog位置（GTID）的自动恢复。可以对数据进行分区以解决数据倾斜问题。发送到 Kafka 的数据支持数据库、表和列。等级数据分区的工作方式是伪装成slave，接收binlog事件，然后根据scheme信息进行组装，可以接受ddl、xid、row等各种事件binlog的引入。

　　Binlog是mysql中的二进制日志。主要用于记录mysql数据库中发生或潜在改变数据的SQL语句，以二进制形式存储在磁盘中。如果我们后面需要配置主从数据库，如果我们需要从数据库同步主数据库的内容，我们可以通过binlog进行同步。说白了，可以用binlog来解决mysql数据库中数据的实时同步问题。还有三种 binlog 格式：STATEMENT、ROW 和 MIXED。

　　基于行的复制（RBR）：它不记录每个 SQL 语句的上下文信息。它只需要记录修改了哪些数据，修改成了什么。混合复制（MBR）：以上两种模式的结合。一般复制使用 STATEMENT 方式保存 binlog。对于 STATEMENT 模式下无法复制的操作，使用 ROW 模式保存 binlog。MySQL会根据执行的SQL语句进行选择。日志保存方法。

　　因为语句只有sql，没有数据，无法获取到原来的变更日志，一般建议使用ROW模式）mysql数据实时同步，我们可以通过解析mysql bin-log来实现，有很多bin-log的解析方式，可以通过canal或者max-well实现。下面是各种提取方法的对比介绍

　　启用binlog功能第一步：添加一个普通用户maxwell

　　在mysql中添加一个普通用户maxwell，因为maxwell软件默认用户是maxwell用户，进入mysql客户端，然后执行如下命令授权

　　mysql -uroot  -p 

set global validate_password_policy=LOW;

set global validate_password_length=6;

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

GRANT ALL ON maxwell.* TO 'maxwell'@'%';

GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';

flush privileges;

　　第二步：开启binlog机制

　　对于我们的mysql数据库，我们可以打开mysql的binlog功能，通过修改mysql配置文件来启用。修改配置文件后，我们需要重启mysql服务

　　sudo vim /etc/my.cnf

# 添加或修改以下三行配置

log-bin= /var/lib/mysql/mysql-bin # 日志存放位置

binlog-format=ROW # binlog格式

server_id=1 # server id

# 重启mysql

sudo service mysqld restart

　　第三步：验证mysql服务

　　# 进入mysql客户端，并执行以下命令进行验证

mysql -uroot -p

mysql> show variables like '%log_bin%';

+---------------------------------+-----------------------------+

| Variable_name                   | Value                       |

+---------------------------------+-----------------------------+

| log_bin                         | ON                          |

| log_bin_basename                | /var/lib/mysql/binlog       |

| log_bin_index                   | /var/lib/mysql/binlog.index |

| log_bin_trust_function_creators | OFF                         |

| log_bin_use_v1_row_events       | OFF                         |

| sql_log_bin                     | ON                          |

+---------------------------------+-----------------------------+

root@0f3525c74115:/# cd /var/lib/mysql/

root@0f3525c74115:/var/lib/mysql# ls -la

-rw-r-----  1 mysql mysql      324 Jul  2 14:54  binlog.000007

-rw-r-----  1 mysql mysql      156 Jul  2 14:54  binlog.000008

-rw-r-----  1 mysql mysql       32 Jul  2 14:54  binlog.index

# 以上是binlog文件说明已经开启

　　安装maxwell实现实时采集mysql数据第一步：下载maxwell

　　下载：

　　第二步：修改配置

　　启动服务。启动我们的动物园管理员服务。启动kafka服务并创建kafka主题。启动maxwell服务，测试插入数据到数据库，查看kafka是否可以同步到mysql数据

　　# 创建kafka topic 分区 3 副本 2

bin/kafka-topics.sh  --create --topic maxwell_kafka --partitions 3 --replication-factor 2 --zookeeper node01:2181

# 创建kafka 消费者 

bin/kafka-console-consumer.sh --topic maxwell_kafka --from-beginning --bootstrap-server node01:9092,node02:9092,node03:9092

# 启动maxwell

bin/maxwell

　　插入数据测试

　　-- 创建目标库

CREATE DATABASE /*!32312 IF NOT EXISTS*/`test` /*!40100 DEFAULT CHARACTER SET utf8 */;

USE `test`;

-- 创建表

CREATE TABLE `myuser` (

  `id` int(12) NOT NULL,

  `name` varchar(32) DEFAULT NULL,

  `age` varchar(32) DEFAULT NULL,

  PRIMARY KEY (`id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- 插入数据

insert  into `myuser`(`id`,`name`,`age`) values (1,'zhangsan',NULL),(2,'xxx',NULL),(3,'ggg',NULL),(5,'xxxx',NULL),(8,'skldjlskdf',NULL),(10,'ggggg',NULL),(99,'ttttt',NULL),(114,NULL,NULL),(121,'xxx',NULL);

-- 岂容kafka消费者，会看到消费的数据

bin/kafka-console-consumer.sh --bootstrap-server node01:9092,node02:9092,node03:9092 --topic  maxwell_kafka