Maxwell - 增量数据同步工具

2024-02-20 17:20

本文主要是介绍Maxwell - 增量数据同步工具,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

前言

        今天来学习一个新的大数据小工具 Maxwell ,它和 Sqoop 很像。Sqoop主要用于在 Hadoop (比如 HDFS、Hive、HBase 等)和关系型数据库之间进行数据的批量导入和导出,而 Maxwell 则主要用于监控数据库的变化(通过监控 binlog ),并将变化的数据以JSON格式发布到消息队列(一般是 Kafka)或 Redis 中。

        Sqoop 一般都是在特定时间(比如用 Azkaban 调度在每天0点进行作业),可以用于离线数仓的数据同步问题;但是对于实时数据分析,Sqoop 并不方便。所以这里就引出了 Maxwell 这个工具,它的特点包括轻量级、能够捕获完整的历史数据(通过bootstrap功能)以及支持断点还原(即在错误解决后可以从上次的位置继续读取数据)。然而,Maxwell只支持 json 格式,并且不能直接支持HA(高可用性)。

        这里说起 Sqoop 我又想到了 DataX ,它也是一款全量数据采集工具,和 Sqoop 很相似,但是还是有所区别:

  • Sqoop是基于Hadoop生态系统的,充分利用了MapReduce计算框架进行数据的导入导出。而DataX仅仅在运行DataX的单台机器上进行数据的抽取和加载。
  • Sqoop采用MapReduce框架,具有良好的并发性和容错性,可以在多个节点同时进行import或export操作。而DataX则是单进程的,没有并发性和容错性。
  • Sqoop主要支持在关系型数据库和Hadoop组件(如HDFS、Hive、HBase)之间进行数据迁移,但不支持在Hadoop组件之间或关系型数据库之间进行数据迁移。相比之下,DataX支持更广泛的数据迁移场景,包括关系型数据库、Hadoop组件以及其他数据源之间的数据迁移。
  • Sqoop在导入导出数据时,会生成一个MapReduce作业在Hadoop集群中运行,可以处理大规模的数据,但不具备统计和校验能力。而DataX在传输过程中可以进行数据过滤,并且可以统计传输数据的信息,对于业务场景复杂、表结构变更的情况更为适用。
  • Sqoop采用命令行的方式调用,容易与现有的调度监控方案相结合。而DataX采用XML配置文件的方式,开发运维上相对不太方便。

接下来就来学习 Maxwell ,内容不多,过两天再把 DataX 搞定。

1、MaxWell 简介

1.1 Maxwell概述

        Maxwell 是由美国 Zendesk 公司开源,用 Java 编写的 MySQL 变更数据抓取软件。它会实时监控Mysql数据库的数据变更操作(包括insert、update、delete),并将变更数据以 JSON 格式发送给 Kafka、Kinesi、RabbitMQ、Redis 或者其它平台的应用程序。

        官网地址:Maxwell's Daemon

1.2 Maxwell 输出数据格式

我们知道 Maxwell 是以 JSON 格式传输数据的,上面显示了不同数据库数据操作对应的 json 格式,这里是这些 json 字段的说明:

字段

解释

database

变更数据所属的数据库

table

表更数据所属的表

type

数据变更类型

ts

数据变更发生的时间

xid

事务id

commit

事务提交标志,可用于重新组装事务

data

对于insert类型,表示插入的数据;对于update类型,标识修改之后的数据;对于delete类型,表示删除的数据

old

对于update类型,表示修改之前的数据,只包含变更字段

2、Maxwell 原理

        Maxwell的工作原理是实时读取MySQL数据库的二进制日志(Binlog),从中获取变更数据,再将变更数据以JSON格式发送至Kafka等流处理平台。

2.1 MySQL二进制日志

2.1.1、什么是 binlog

        二进制日志(Binlog)是MySQL服务端非常重要的一种日志,它记录了所有的 DDL 和 DML (除了查询语句)语句,以事件的形式记录,还包含所执行的消耗的时间,MySQL 的二进制日志是事务安全类型的。Maxwell的工作原理和主从复制密切相关。

        一般开启二进制日志会有 1% 的性能损耗。二进制日志有两个最重要的场景:

  • MySQL Replication 在 Master 端开启 binlog,Master 把它的二进制日志传递给其它 salves 来达到 master-salve 数据一致的目的。
  • 通过使用 musqlbinlog 工具来进行数据恢复。

        MySQL二进制日志的存储机制很像我们之前学的Kafka 文件存储机制,它也包含两类文件:二进制索引文件(存储二进制日志文件索引,后缀为 .index)和二进制日志文件(用于记录所有 DDL 和 DML 语句,后缀为 .00000*)。

2.1.2、binlog 的开启

  • 找到 MySQL 配置文件的位置
  • linux:/etc/my.cnf (可以通过 locate my.cnf) 查找位置
  • window:\my.ini
  • 修改配置:在 [mysqlId] 区块,设置添加 log-bin=mysql-bin (这个表示 binlog 的前缀是 mysql-bin,也可以换成别的,也就是说以后生成的日志文件就是 mysq-bin.000001 格式的文件,每次 mysql 重启或者达到单个文件大小的阈值时,生成一个新的日志文件、按顺序编号)

2.1.3、binlog 的分类设置

mysql binlog 的格式共有三种:STATEMENT,MINED,ROW。

在配置文件中可以选择配置:binllog_format = statment|mixed|row

  • statement:语句级,它会记录每一次执行写操作的雨具。相比较 row 模式节省空间,但是可能产生不一致,比如 update student enroll_time = now(); 这种 SQL 如果进行数据恢复那一定是有问题的。
    • 优点:节省空间
    • 缺点:有可能造成数据不一致
  • row:行级,binlog 记录每次操作后每行记录的变化,相当于它记录的是我们每次操作后表格的内容。
    • 优点:保持数据的绝对一致性,不管什么sql,它只记录执行后的结果
    • 缺点:占用磁盘较大
  • mixed:混合级别,statement 的升级版,一定程度上解决了 statement 模式因为一些情况而造成的数据不一致问题。比如对于 SQL 中包含随机数或者时间的语句,它保存的是 SQL 执行的结果;对于 SQL 不包含随机数或者时间的语句,它保存的是 SQL 语句本身。
    • 优点:节省空间
    • 缺点:极个别情况下仍然存在不一致

综上,Maxwell 做监控分析,选择 row 模式比较合适(必须选择 row),因为我们的下游(比如 Kafka 是无法执行 sql 语句的)

2.1.4、Maxwell 和 Canal 对比

对比CanalMaxwell
语言

java

java
数据格式自由json
采集模式增量全量/增量
数据落地定制支持 kafka 等多种平台
HA支持支持

Canal 的原理和 Maxwell 都是通过监控 binlog 实现的。 

2.2 MySQL主从复制

        MySQL的主从复制,就是用来建立一个和主数据库完全一样的数据库环境,这个数据库称为从数据库。

1)主从复制的应用场景如下:

(1)做数据库的热备:主数据库服务器故障后,可切换到从数据库继续工作。

(2)读写分离:主数据库只负责业务数据的写入操作,而多个从数据库只负责业务数据的查询工作,在读多写少场景下,可以提高数据库工作效率。

2)主从复制的工作原理如下:

(1)Master主库将数据变更记录,写到二进制日志(binary log)中

(2)Slave从库向mysql master发送dump协议,将master主库的binary log events拷贝到它的中继日志(relay log)

(3)Slave从库读取并执行中继日志中的事件,将改变的数据同步到自己的数据库。

2.3 Maxwell原理

        Maxwell 的原理很简单,就是将自己伪装成slave,并遵循MySQL主从复制的协议,从master同步数据。

3、Maxwell 部署

3.1、启用 MySQL binlog

MySQL服务器的Binlog默认是未开启的,如需进行同步,需要先进行开启。

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

 sudo vim /etc/my.cnf

2)增加如下配置

[mysqld]#数据库id
server-id = 1
#启动binlog,该参数的值会作为binlog的文件名
log-bin=mysql-bin
#binlog类型,maxwell要求为row类型
binlog_format=row
#启用binlog的数据库,需根据实际情况作出修改
binlog-do-db=gmall
#如果需要监控多个数据库 不要加逗号 直接新开一行
binlog-do-db=test

如果我们要监控除了个别数据库之外的其它所有数据库,我们可以使用 binlog-ignore-db 属性。

3)重启MySQL服务

sudo systemctl restart mysqld

4)查看是否修改完成

show variables like '%binlog%';

我们看到 binlog_format 的 Value 现在是 ROW ,说明 binlog 配置成功。

5)查看 binlog 文件

进入 /var/lib/mysql 目录,查看 MySQL 生成的 binlog 文件:

cd /var/lib/mysql
sudo ls -l | grep mysql-bin

 注:MySQL 生成的 binlog 文件初始大小一定是 514 字节。

3.2、初始化 maxwell 元数据库

(1)创建元数据库 maxwell 用于存储 maxwell 的元数据

create database maxwell;

 (2)设置 mysql 用户密码安全级别

set global validate_password_policy=0;
set global validate_password_length=4;

如果上面的命令报错,去 /etc/my.cnf 的 [mysqld] 下添加下面两行:

plugin-load-add=validate_password.so
validate-password=FORCE_PLUS_PERMANENT

然后重启 mysqld 服务

sudo systemctl restart mysqld

(3)分配一个用户账号可以操作该元数据库

GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY '123456';

(4)分配这个账号可以监控其他数据库的权限

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

(5)刷新 mysql 表权限

flush privileges;

3.3、Maxwell 进程启动

3.3.1、使用命令行参数启动

bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout
  • user:连接 mysql 的用户(上面我们已经创建了)
  • password:连接 mysql 的用户密码
  • host: mysql 安装的主机名
  • producer:生产者模式(stdout:控制台,kafka:kafka 集群)

查看 maxwell 输出: 

 我们可以看到,我们的 insert 语句被 maxwell 转为了 json 被输出到了控制台。

3.3.2、定制化配置文件启动

 编辑配置文件 config.properties:

log_level=infoproducer=stdout
kafka.bootstrap.servers=localhost:9092# mysql login info
host=hadoop102
user=maxwell
password=123456

 启动 Maxwell:

4、Maxwell 使用

4.1、监控 MySQL 数据并在控制台打印

        上面我们已经演示过了,也就是首先在 mysql 的配置文件中指定需要打印 binlog 的数据库,然后通过命令或者配置文件来开启 Maxwell 进程:

bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout

接着我们对被监听的数据库的操作信息就会以 json 的格式打印出来: 

4.2、监控 MySQL 数据并输出到 Kafka

        监控 MySQL 数据并输出到 Kafka,这种需求是我们实际工作用的最多的,这也是我们离线数仓和实时数仓所必须的。

4.2.1、实例

1)启动 Zookeeper 和 Kafka

2)启动 Maxwell 监控 binlog

bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=kafka --kafka.bootstrap.servers=hadoop102:9092 --kafka_topic=maxwell

注:这里我们制定了 Kafka 集群地址为 hadoop102:9092 ,主题为 maxwell,虽然主题并不存在,但是 kafka 会自动创建我们指定的主题

3)在 hadoop103 消费 maxwell 主题

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

4)在 MySQL 中增加一条数据再删除: 

 可以看到我们的数据成功被 Kafka 消费,说明业务数据日志成功传输到 Kafka 。

4.2.2、Kafak 主题数据的分区控制

        上面的案例中,我们不论 MySQL 的那个数据库,哪张表,都被保存到了 Kafka 的 maxwell 主题,这样显然很乱。所以这里我们需要解决一下:
        在生产环境中,我们一般都会用 maxwell 监控多个 mysql 库的数据,然后将这些数据发往 Kafka 的一个主题,并且这个主题肯定是多分区的,为了提高并发度。那么如何控制这些数据的分区问题,那就变得至关重要,实现步骤如下:
(1)修改 maxwell 的配置文件,定制化启动 maxwell 进程

log_level=infoproducer=kafka
kafka.bootstrap.servers=hadoop102:9092# mysql login info
host=hadoop102
user=maxwell
password=123456# kafka 主题
kafka_topic=maxwell3# 默认配好的
kafka.compression.type=snappy
kafka.retries=0
kafka.acks=1#producer_partition_by=database # [database, table, primary_key, transaction_id, thread_id, column]
# 常用的按照库分区、按照表分区,这里我们测试按照库分区
producer_partition_by=database

(2)手动创建 Kafka 多分区主题

kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 --create --replication-factor 2 --partitions 3 --topic maxwell3

 查看主题信息:

这样,我们不同数据库的操作记录就会保存到 Kafka 不同的分区当中, 分区可以使得集群负载均衡,还可以提高提高读写效率。

4.3、监控 MySQL 指定表输出控制台

(1)运行 maxwell 来监控 mysql 指定表数据更新

监控 test 库下的 staff 表

bin/maxwell --user='maxwell' --password='123456' --host='hadoop102' --producer=stdout --filter 'exclude: *.*,include:test.staff'

(2)向 staff 表中插入数据,查看 maxwell 控制台输出

(3)向 ws2 表中插入一条数据,查看

注:表 ws2 并不是我们指定监控的表

结果:maxwell 控制台并没有任何输出 

注:还可以设置 include.test.* 来表示监控 test 库下的所有表

4.4、监控 MySQL 指定表全量输出到控制台

        Maxwell 进程默认只能监控 mysql 的 binlog 日志的新增以及变化的,但是 Maxwell 是支持数据初始化的,可以通过修改 Maxwell 的元数据来对 MySQL 中的某张表进行数据初始化,也就是我们常说的全量同步。具体操作如下:

        需求:将 test 库下的 staff 表中的数据全量导入到 maxwell 控制台打印。

我们可以在数据库 maxwell (这个数据库是我们初始化 maxwell 元数据库时创建的)中查看元数据表 positions 的内容: 

这里的 server_id 是我们在 mysql 的配置文件 /etc/my.cnf 中配置的;这里的 binlog_file 代表的是最新的 binlog 文件 mysql-bin.000003 ;这里的 binlog_position 的值为 2802 代表目前已经读到了第 2802 个字节的位置,意思就说我们现在是从这个偏移量之后开始监听的。

注:也可以通过 SQL:show master status; 来查看当前的 binlog 状态。

1. 向元数据表 maxwell.bootstrap 中插入数据

insert into maxwell.bootstrap(database_name,table_name) values('test','staff');

这就相当于告诉 maxwell,下一次启动 maxwell 作业时,就会调用一个初始化作业,把我们指定的这个 test 数据库下的 staff 表进行一次增量同步。

我们可以看到,通过 bootstrap 表插入数据来完成初始化,这样插入的数据是 json 中 type 字段的值是 "bootstrap-insert" 而不是 "insert"。

再次查看 bootstrap 表: 

我们可以发现,is_complete 字段的值从默认值 0 变成了 1,inserted_rows 从 0 变成了 6,后面的 started_at 和 completed_at 字段的值也不再为空了。

2. 使用 maxwell-bootstrap 脚本工具

除了上面的直接向 maxwell 元数据库中的 bootstrap 表插入数据,也可以使用 maxwell 提供的 maxwell-bootstrap 脚本,本质其实是一样的,具体看你觉得哪个省事。

bin/maxwell-bootstrap --database gmall --table user_info --config ./config.properties

这样,这个初始化作业就算完成了,当我们下次打开 maxwell 进程时,它并不会再次执行这个初始化任务,因为它已经完成过了。

总结

        至此,Maxwell 的学习也已经结束了,待会吧离线数仓项目里的 maxwell 部分完成。总体来说,这个工具还是很好用且简单的,希望有一天自己也可以用屎山堆砌出这么一个自己的框架!

这篇关于Maxwell - 增量数据同步工具的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/728956

相关文章

大模型研发全揭秘:客服工单数据标注的完整攻略

在人工智能(AI)领域,数据标注是模型训练过程中至关重要的一步。无论你是新手还是有经验的从业者,掌握数据标注的技术细节和常见问题的解决方案都能为你的AI项目增添不少价值。在电信运营商的客服系统中,工单数据是客户问题和解决方案的重要记录。通过对这些工单数据进行有效标注,不仅能够帮助提升客服自动化系统的智能化水平,还能优化客户服务流程,提高客户满意度。本文将详细介绍如何在电信运营商客服工单的背景下进行

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

服务器集群同步时间手记

1.时间服务器配置(必须root用户) (1)检查ntp是否安装 [root@node1 桌面]# rpm -qa|grep ntpntp-4.2.6p5-10.el6.centos.x86_64fontpackages-filesystem-1.41-1.1.el6.noarchntpdate-4.2.6p5-10.el6.centos.x86_64 (2)修改ntp配置文件 [r

关于数据埋点,你需要了解这些基本知识

产品汪每天都在和数据打交道,你知道数据来自哪里吗? 移动app端内的用户行为数据大多来自埋点,了解一些埋点知识,能和数据分析师、技术侃大山,参与到前期的数据采集,更重要是让最终的埋点数据能为我所用,否则可怜巴巴等上几个月是常有的事。   埋点类型 根据埋点方式,可以区分为: 手动埋点半自动埋点全自动埋点 秉承“任何事物都有两面性”的道理:自动程度高的,能解决通用统计,便于统一化管理,但个性化定

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

异构存储(冷热数据分离)

异构存储主要解决不同的数据,存储在不同类型的硬盘中,达到最佳性能的问题。 异构存储Shell操作 (1)查看当前有哪些存储策略可以用 [lytfly@hadoop102 hadoop-3.1.4]$ hdfs storagepolicies -listPolicies (2)为指定路径(数据存储目录)设置指定的存储策略 hdfs storagepolicies -setStoragePo

Hadoop集群数据均衡之磁盘间数据均衡

生产环境,由于硬盘空间不足,往往需要增加一块硬盘。刚加载的硬盘没有数据时,可以执行磁盘数据均衡命令。(Hadoop3.x新特性) plan后面带的节点的名字必须是已经存在的,并且是需要均衡的节点。 如果节点不存在,会报如下错误: 如果节点只有一个硬盘的话,不会创建均衡计划: (1)生成均衡计划 hdfs diskbalancer -plan hadoop102 (2)执行均衡计划 hd

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯,先赞后看哦~🎈🎈 🏆 作者简介:景天科技苑 🏆《头衔》:大厂架构师,华为云开发者社区专家博主,阿里云开发者社区专家博主,CSDN全栈领域优质创作者,掘金优秀博主,51CTO博客专家等。 🏆《博客》:Python全栈,前后端开发,小程序开发,人工智能,js逆向,App逆向,网络系统安全,数据分析,Django,fastapi

高效录音转文字:2024年四大工具精选!

在快节奏的工作生活中,能够快速将录音转换成文字是一项非常实用的能力。特别是在需要记录会议纪要、讲座内容或者是采访素材的时候,一款优秀的在线录音转文字工具能派上大用场。以下推荐几个好用的录音转文字工具! 365在线转文字 直达链接:https://www.pdf365.cn/ 365在线转文字是一款提供在线录音转文字服务的工具,它以其高效、便捷的特点受到用户的青睐。用户无需下载安装任何软件,只

烟火目标检测数据集 7800张 烟火检测 带标注 voc yolo

一个包含7800张带标注图像的数据集,专门用于烟火目标检测,是一个非常有价值的资源,尤其对于那些致力于公共安全、事件管理和烟花表演监控等领域的人士而言。下面是对此数据集的一个详细介绍: 数据集名称:烟火目标检测数据集 数据集规模: 图片数量:7800张类别:主要包含烟火类目标,可能还包括其他相关类别,如烟火发射装置、背景等。格式:图像文件通常为JPEG或PNG格式;标注文件可能为X