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 部分完成。总体来说,这个工具还是很好用且简单的,希望有一天自己也可以用屎山堆砌出这么一个自己的框架!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/691774.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

基于MATLAB的QPSK调制解调仿真(仿真图超多,结果超清晰)

基于MATLAB的QPSK调制解调仿真(包含中间各个环节的时域波形、功率谱、频谱图、星座图和眼图、理论与仿真的误码率曲线) 目录 前言 一、QPSK调制解调过程 二、仿真结果 Ⅰ、码元信噪比eb/n010dB时 1、双极性不归零基带信号 2、成型滤波(根升余弦滤波)后 3、调制后 4、加入…

Linux CentOS stream 9 firewalld

随着互联网行业快速发展,服务器成为用户部署网络业务重要的网络工具,但随之而来的就是更密集的网络攻击,这给网站带来了很大的阻碍。防火墙作为保障网络安全的主要设备,可以很好的抵御网络攻击。 防火墙基本上使用硬件和软件两种…

深度学习(16)--基于经典网络架构resnet训练图像分类模型

目录 一.项目介绍 二.项目流程详解 2.1.引入所需的工具包 2.2.数据读取和预处理 2.3.加载resnet152模型 2.4.初始化模型 2.5.设置需要更新的参数 2.6.训练模块设置 2.7.再次训练所有层 2.8.测试网络效果 三.完整代码 一.项目介绍 使用PyTorch工具包调用经典网络架构…

PCIe学习笔记(2)错误处理和AER/DPC功能

文章目录 PCIe ErrorAER (Advanced Error Reporting)DPC (Downstream Port Containment) 处理器上错误通常可分为detected和undetected error。Undetected errors可能变得良性(benign),也可能导致系统故障如silent data corruptions (SDC)。Detected errors则又可分…

MySQL基础学习

MySQL基础 注意:本文的图片截图自尚硅谷MySQL笔记。 一:基本概述: 什么是数据库: 数据库是一种用来存储和管理数据的系统。它是一个组织化的数据集合,可以通过计算机系统进行访问、管理和更新。数据库可以存储各种…

如何使用CloakQuest3r获取受安全服务保护的网站真实IP地址

关于CloakQuest3r CloakQuest3r是一款功能强大的纯Python工具,该工具可以帮助广大研究人员获取和查看受Cloudflare和其他安全服务商保护的网站真实IP地址。 Cloudflare是一种广泛采用的网络安全和性能增强服务,而CloakQuest3r的核心任务就是准确识别隐…

怎么在抖音带自己的货?带货方式和带货要求,如下所示

我是王路飞。 不管你是无货源的新手小白,还是有货源的厂家/供应链,想在抖音卖货的话,无非就两种方式:要么开店、要么开直播带货。 看似都是在抖音卖货,但其实这是两条不同的赛道。 这篇文章就给你们聊下想在抖音卖货…

【Redis】理论进阶篇------Redis的持久化

一、前言 前面学习了Redis的相关的十大数据类型以及用SpringBoot集成我们的Redis的工具代码的书写。从这篇文章开始,就会从Redis相关的一些理论(也是面试和工作的热点知识)如:Redis的持久化、Redis的订阅发布模型、Redis集群环境搭…

用python绘制黄金价格变化曲线

首先你得从mt4把数据导出为csv:mt4如何导出数据-CSDN博客 1、引入必要的库 import numpy as np import pandas as pd import matplotlib.pyplot as plt 2、然后通过pandas载入csv数据 raw pd.read_csv("XAUUSDm1.csv", headerNone, index_colNone, p…

P1024 [NOIP2001 提高组] 一元三次方程求解题解

题目 有形如:这样的一个一元三次方程。给出该方程中各项的系数(a,b,c,d均为实数),并约定该方程存在三个不同实根(根的范围在−100至100之间),且根与根之差的绝对值≥1。要求由小到大依次在同一…

文件包含+文件上传漏洞(图片马绕过)

目录 一.文件包含二.文件上传三.图片马四.题目 一.文件包含 将已有的代码以文件形式包含到某个指定的代码中,从而使用其中的代码或者数据,一般是为了方便直接调用所需文件,文件包含的存在使得开发变得更加灵活和方便(若对用户输入…

网络安全-pikachu之文件上传漏洞1

文件上传漏洞是危害极大的,一旦成功,可以获取服务器的最高权限。 pikachu介绍: 文件上传功能在web应用系统很常见,比如很多网站注册的时候需要上传头像、上传附件等等。当用户点击上传按钮后,后台会对上传的文件进行判…

☀️将大华摄像头画面接入Unity 【1】配置硬件和初始化摄像头

一、硬件准备 目前的设想是后期采用网口供电的形式把画面传出来,所以这边我除了大华摄像头还准备了POE供电交换机,为了方便索性都用大华的了,然后全都连接电脑主机即可。 二、软件准备 这边初始化摄像头需要用到大华的Configtool软件&#…

【知识点】CNN中concat与add的区别

cat操作经常用于将特征联合,多个卷积特征提取框架提取的特征融合或者是将输出层的信息进行融合;而add层更像是信息之间的叠加。 add是在一个特征上增加其语义信息,对最终的图像的分类是有益;cat导致的结果改进可能是由于cat操作通…

工业自动化部署选择主板的关键因素

**在构建任何计算机时,选择合适的主板至关重要。**对于游戏台式机,您需要选择能够支持您玩的游戏类型而不会出现任何问题的最新和最佳规格。当涉及工业应用时,影响您决策的变量变得更加重要。作为任何基于计算的应用中最关键的组件之一&#…

搜维尔科技:分析OptiTrack光学动作捕捉应用领域!

虚拟制作 当今虚拟制作阶段低延迟、超精确摄像机跟踪的事实上的标准。 用于运动科学的 OptiTrack OptiTrack 系统提供世界领先的测量精度和简单易用的工作流程,为研究人员和生物力学师的研究提供理想的 3D 跟踪数据。对所有主要数字测力台、EMG 和模拟设备的本机即…

trojan 突然无法上网

[ERROR] 2024/02/19 18:14:45 github.com/p4gefau1t/trojan-go/tunnel/tls.(*Server).acceptLoop.func1:server.go:140 tls handshake failed | remote error: tls: bad certificate 报证书问题,更新证书发现无法解决 最后突然客户端有一个配置 验证证书&#xf…

佳能2580的下载手册

凡是和电子产品有关的产品其内部都开始不断地进行内卷,在不断地内卷背后,意味着科技更新和换代,自己也入手了一台佳能2580的打印机,一台相对比较老式的打印机,以此不断地自己想要进行打印的需要。 下载的基础步骤&…

【ansible】认识ansible,了解常用的模块

目录 一、ansible是什么? 二、ansible的特点? 三、ansible与其他运维工具的对比 四、ansible的环境部署 第一步:配置主机清单 第二步:完成密钥对免密登录 五、ansible基于命令行完成常用的模块学习 模块1:comma…

Bert基础(二)--多头注意力

多头注意力 顾名思义,多头注意力是指我们可以使用多个注意力头,而不是只用一个。也就是说,我们可以应用在上篇中学习的计算注意力矩阵Z的方法,来求得多个注意力矩阵。让我们通过一个例子来理解多头注意力层的作用。以All is well…