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 字段的值也不再为空了。

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

总结

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

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

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

相关文章

如何使用Docker部署Drupal并结合cpolar实现固定公网地址访问

文章目录 前言1. Docker安装Drupal2. 本地局域网访问3 . Linux 安装cpolar4. 配置Drupal公网访问地址5. 公网远程访问Drupal6. 固定Drupal 公网地址 前言 Dupal是一个强大的CMS,适用于各种不同的网站项目,从小型个人博客到大型企业级门户网站。它的学习…

2024年及以后在您的项目中使用的最佳CSS框架

在过去几年中,CSS已经取得了长足的进步。在过去,您可能会使用CSS来创建依赖于HTML表格和CSS浮动作为其布局系统的简单外观的Web应用程序。而现在,您可以设计复杂的交互式用户界面,具有优雅的设计。 尽管CSS变得越来越先进&#x…

虚拟机centos7 网络IP冲突

修改其中一个虚拟机IP 1: 设置虚拟机网络配置器的模式为NAT模式,操作方式如下图所示 2:点击虚拟网络编辑器 3:点击NAT设置 4:点击DHCP配置 5: 修改配置文件来指定IP并可以连接到外网,在roo…

【通讯录案例-数据存储总结 Objective-C语言】

一、我们简单的把“数据存储”总结一下, 1.上午,我们说的三种存储方式, 1)plist 2)Preference(偏好设置) 3)NSKeyedArchiver(归档、解档) 三种存储方式, 2.什么能做plist存储, 1)writeToFile:方法, 实际上,这个东西,才是关键, 有一个对象,可以去写w…

ETL数据集成工具DataX、Kettle、ETLCloud特点对比

ETL数据集成工具 对于数据仓库,大数据集成类应用,通常会采用ETL工具辅助完成。ETL,是英文 Extract-Transform-Load 的缩写,用来描述将数据从来源端经过抽取(extract) 、交互转换(transform) 、加载(load)至的端的过程当前的很多应…

spring @Transactional注解参数详解

事物注解方式: Transactional 当标于类前时, 标示类中所有方法都进行事物处理 , 例子: 1 Transactional public class TestServiceBean implements TestService {}当类中某些方法不需要事物时: Transactional public class TestServiceBean implements TestService {private…

Vue24 收集表单数据 实例

实例 <!DOCTYPE html> <html><head><meta charset"UTF-8" /><title>收集表单数据</title><script type"text/javascript" src"../js/vue.js"></script></head><body><!-- 收集…

SpringCloud-Feign:负载均衡(基于服务端)

7.Feign&#xff1a;负载均衡(基于服务端) 7.1 Feign简介 Feign是一个开源的声明式HTTP客户端&#xff0c;它可以简化HTTP API的调用过程。Feign的设计目标是使得使用者可以像调用本地方法一样调用远程服务&#xff0c;使得编写和维护HTTP客户端变得更加简单。类似controller…

2045第六题 拯救花园 (flowers)

题目大意&#xff1a; 有n只兔子&#xff0c;每只兔子抓回去的时间为ti,回来的时间也是ti,则抓一只兔子要2*ti的时间&#xff0c;di则为每只兔子一个时间单位能吃多少草&#xff0c;用最优方法做的话它们一共吃了多少草&#xff08;最少&#xff09; 贪心标准&#xff1a; 我们…

[深度学习] 卷积神经网络“卷“在哪里?

​ &#x1f308; 博客个人主页&#xff1a;Chris在Coding &#x1f3a5; 本文所属专栏&#xff1a;[深度学习] ❤️ 热门学习专栏&#xff1a;[Linux学习] ⏰ 我们仍在旅途 目录 1.卷积的定义 2.卷积的"卷"在哪里 3.什么又是卷积神…

C++学习Day06之多继承基本语法

目录 一、程序及输出二、分析与总结 一、程序及输出 多继承的基本语法如下&#xff1a; class Base1 { // Base1 class definition }; class Base2 { // Base2 class definition }; class Derived : public Base1, protected Base2 { // Derived class definition }; #include…

OpenAI划时代大模型——文本生成视频模型Sora作品欣赏(一)

Sora介绍 Sora是一个能以文本描述生成视频的人工智能模型&#xff0c;由美国人工智能研究机构OpenAI开发。 Sora这一名称源于日文“空”&#xff08;そら sora&#xff09;&#xff0c;即天空之意&#xff0c;以示其无限的创造潜力。其背后的技术是在OpenAI的文本到图像生成模…

MySQL-锁(LOCK)

文章目录 1. 锁是什么&#xff1f;2. 全局锁2.1 相关语法2.2 特点 3. 表级锁3.1 表锁3.1.1 共享读锁&#xff08;S&#xff09;3.1.2 排它写锁&#xff08;X&#xff09; 3.2 元数据锁&#xff08;MDL&#xff09;3.2 意向锁&#xff08;IS、IX&#xff09; 4. 行级锁4.1 行锁 …

Python学习(16)|列表_遍历_排序_max_min_sum

列表的遍历&#xff1a; a [10,20,30,40] for obj in a: #obj 是临时变量名称&#xff0c;随意起名print(obj) 执行结果&#xff1a; 复制列表所有的元素到新列表对象&#xff1a; list1 [30,40,50] list2 list1 #只是将list2也指向了列表对象。也就是说list…

Springboot返回给前端的日期变成时间戳问题处理

问题&#xff1a; 解决方案 在对应的Entity类中&#xff0c;给对应属性添加JsonFormat​注解 ​JsonFormat​注解用于将Date日期格式化为指定格式的字符串。由于在序列化时间时是按照国际标准时间GMT进行格式化的&#xff0c;最后接受到的数据会早8个小时&#xff0c;所以应该…

CF778A String Game 题解

文章目录 CF778A String Game 题解题面翻译Input DataOutput DataInput Sample 1Output Sample 1题目描述输入格式输出格式样例 #1样例输入 #1样例输出 #1 样例 #2样例输入 #2样例输出 #2 提示算法&#xff1a;二分代码&#xff1a; CF778A String Game 题解 link 题面翻译 …

【白嫖8k买的机构vip教程】安卓设备连接电脑进行adb命令操作

Android 真机进行adb 命令的操作&#xff1a; 连接方式 &#xff1a; 数据线连接&#xff1a; 1&#xff09;手机需通过数据线连接电脑&#xff1b; 2&#xff09;手机调出开发者选项&#xff1b;手机操作步骤&#xff1a;设置——系统——关于手机&#xff08;平板电脑&am…

CDN缓存有什么作用?

CDN缓存是内容分发网络的核心技术之一&#xff0c;它的作用在于通过将内容缓存在边缘服务器上&#xff0c;提高内容的访问速度和可用性。以下是CDN缓存的几个主要作用&#xff1a; 加速内容的访问速度 CDN缓存通过将内容缓存在距离用户更近的边缘服务器上&#xff0c;减少了内…

【办公类-16-07-02】“2023下学期 周计划-户外游戏 每班1周五天相同场地,6周一次循环”(python 排班表系列)

背景需求&#xff1a; 又到了开学季&#xff0c;新的自主游戏&#xff08;户外游戏&#xff09;安排表出炉了。 这张是贴在美术活动室的安排表&#xff0c;我需要转换成班级为单位的安排表&#xff0c;便于批量制作周计划。 设计思路&#xff1a; 1、一个班级每周轮到的一个场…

电商+支付双系统项目------怎么用代码实现支付系统?

前言 上篇文章讲了我们应该怎么设计支付系统&#xff0c;构建了一个支付系统的蓝图&#xff0c;这篇文章我来讲一下怎么用代码来设计支付系统。当然&#xff0c;我肯定不是手把手的教你哈哈哈&#xff0c;我只是把支付系统的核心的部分展示给大家看&#xff0c;让大家了解一下…