【数仓】Maxwell软件安装及配置,采集mysql数据

相关文章

  • 【数仓】基本概念、知识普及、核心技术
  • 【数仓】数据分层概念以及相关逻辑
  • 【数仓】Hadoop软件安装及使用(集群配置)
  • 【数仓】Hadoop集群配置常用参数说明
  • 【数仓】zookeeper软件安装及集群配置
  • 【数仓】kafka软件安装及集群配置
  • 【数仓】flume软件安装及配置
  • 【数仓】flume常见配置总结,以及示例

一、Maxwell有什么作用

Maxwell是一个读取MySQL binlogs(二进制日志)的应用程序,它的主要作用是将MySQL中的变更数据(包括insert、update、delete等操作)实时捕获,并将这些变更数据以JSON格式发送到Kafka、Kinesis、RabbitMQ等流数据处理平台。通过这种方式,Maxwell实现了数据的实时同步和变更捕获,为数据仓库提供了实时、准确的数据源。

具体来说,Maxwell可以支持SELECT * FROM table的方式进行全量数据初始化,同时还支持在主库发生failover后自动恢复binlog位置(GTID)。此外,Maxwell还可以对数据进行分区,解决数据倾斜问题,并支持database、table、column等级别的数据分区。在工作方式上,Maxwell伪装为MySQL的Slave,接收binlog events,然后根据schemas信息拼装,可以接受ddl、xid、row等各种event。

总的来说,在数仓中,Maxwell的作用主要是实时捕获MySQL中的变更数据,并将这些数据以流的形式发送到下游处理平台,从而实现数据的实时同步和处理。这对于需要实时分析、处理大量数据的场景(如实时报表、实时推荐系统等)非常有用。

二、环境准备

准备1台虚拟机

  • Hadoop131:192.168.56.131

本例系统版本 CentOS-7.8,已安装jdk1.8

关闭防火墙

systemctl stop firewalld

三、Maxwell安装配置

1、Maxwell下载安装

这里使用的是1.29.2版本,1.30.0以上的版本需要jdk11

# 下载解压
wget --no-check-certificate https://github.com/zendesk/maxwell/releases/download/v1.29.2/maxwell-1.29.2.tar.gz
tar -xzvf maxwell-1.29.2.tar.gz
mv  maxwell-1.29.2/ /data/maxwell/

2、开启mysql Binlog模式

在 MySQL 8.0 中,要开启二进制日志(binlog),你需要在 MySQL 的配置文件(my.inimy.cnf)中设置相应的选项。以下是如何开启 binlog 的步骤:

  1. 找到配置文件:首先,找到 MySQL 的配置文件。在 Windows 上,它通常位于 MySQL 安装目录下的 my.ini 文件中。在 Linux 或其他 Unix-like 系统上,它通常位于 /etc/my.cnf/etc/mysql/my.cnf。如果找不到这些文件,请根据你的安装方式和操作系统查找正确的位置。

  2. 编辑配置文件:使用文本编辑器打开配置文件。在 [mysqld] 部分下添加或修改以下行:

    [mysqld]
    log-bin=mysql-bin  # 开启 binlog 并设置日志文件名前缀为 mysql-bin
    server-id=1        # 设置服务器 ID,确保每个 MySQL 实例的 server-id 是唯一的
    binlog-format=row
    binlog-do-db=user  # 指定需要记录二进制日志的数据库名
    # 如果需要配置多个数据库,可以写多行,类似如下
    # binlog-do-db=database1  
    # binlog-do-db=database2
    

    log-bin 选项用于启用二进制日志并设置日志文件的名称前缀。server-id 是必需的,用于在复制环境中标识每个 MySQL 服务器实例。

  3. 重启 MySQL 服务:保存配置文件后,需要重启 MySQL 服务以使更改生效。根据你的操作系统和安装方式,使用相应的命令来重启 MySQL 服务。例如,在 Linux 上,你可能会使用 sudo systemctl restart mysqlsudo service mysql restart。在 Windows 上,你可以通过服务管理器重启 MySQL 服务。

  4. 验证 binlog 是否已开启:重启 MySQL 服务后,登录到 MySQL 服务器并运行以下 SQL 查询来验证 binlog 是否已开启:

    SHOW VARIABLES LIKE 'log_bin';
    

    如果返回的结果是 ON,则表示 binlog 已成功开启。

请注意,开启 binlog 可能会增加一些磁盘 I/O 开销,因为所有更改数据的查询都会被记录到日志文件中。确保你的磁盘有足够的空间来存储这些日志文件,并定期监控和管理它们的大小和数量。

3、创建Maxwell所需的数据库和用户

# 创建数据库
CREATE DATABASE maxwell;# 创建 maxwell 用户并赋予其必要权限
CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
GRANT ALL ON maxwell.* to 'maxwell'@'%';
GRANT SELECT,REPLICATION CLIENT,REPLICATION SLAVE ON *.* TO 'maxwell'@'%'

4、配置Maxwell属性

重命名根目录(/data/maxwell)下的 config.properties.example,然后修改配置

mv config.properties.example config.properties

vim config.properties,如下配置:

producer=kafka
#目标 Kafka 集群地址
kafka.bootstrap.servers=hadoop131:9092,hadoop132:9092,hadoop133:9092
#目标 Kafka topic,可静态配置,例如:maxwell,也可动态配置,例如:%{database}_%{table}
kafka_topic=topic_db# mysql login info
host=192.168.56.1
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true#过滤目标数据库中无须采集的表
filter=exclude:gmall.sys_log
#指定数据按照主键分组进入 Kafka 不同分区,避免数据倾斜
producer_partition_by=primary_key

5、启动Maxwell

1)创建Maxwell启动脚本mxw.sh

vi /usr/bin/mxw.sh
# 修改文件权限
chmod 777 /usr/bin/mxw.sh

2)复制如下内容

#!/bin/bashMAXWELL_HOME=/data/maxwell#1. 判断参数个数
if [ $# -lt 1 ]
thenecho Not Enough Arguement!exit;
fistatus_maxwell(){result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`return result
}start_maxwell(){status_maxwellif [[ $? -lt 1 ]]; thenecho "启动 Maxwell"$MAXWELL_HOME/bin/Maxwell --config $MAXWELL_HOME/config.properties --daemonelseecho "Maxwell 正在运行"fi
}stop_maxwell(){status_maxwellif [[ $? -gt 0 ]]; thenecho "停止 Maxwell"ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep |awk '{print $2}' | xargs kill -9elseecho "Maxwell 未在运行"fi
}case $1 in"start")start_maxwell;;"stop")stop_maxwell;;"restart")stop_maxwellstart_maxwell;;*)echo "Input Args Error...";;
esac

3)执行脚本 mxw.sh start 启动

先要启动zookeeper 和 kafka

6、测试

经过上面的步骤,已经启动了zookeeper 、 kafka、和maxwell

链接数据库,随便修改一个表的数据,比如我修改的是user.t_user表。

链接kafka,可以看见自动创建了topic topic_db,查看同步数据:

{"database": "user","table": "t_user","type": "update","ts": 1709895217,"xid": 3760,"commit": true,"data": {"id": 28,"name": "1232","msg": "新用户注册事务提交事件","create_time": "2023-05-20 23:59:59","last_login_time": null,"status": 1},"old": {"name": "123"}
}

经过上面测试,说明使用Maxwell,监控mysql binlog 同步数据的操作已经打通!

四、Maxwell 全量同步

1、使用 maxwell-boostrap 全量同步数据

/data/maxwell/bin/maxwell-bootstrap --database user --table t_user --config /data/maxwell/config.properties

主要参数:

  • --database dbname:指定要从中获取初始数据快照的数据库名称。
  • --table tablename:指定要从中获取初始数据快照的表名称。这个参数是可选的,如果省略,则默认会同步整个数据库的所有表。但是,请注意,根据 Maxwell 的具体版本和配置,这个行为可能会有所不同。
  • --config /data/maxwell/config.properties:指定 Maxwell 的配置文件路径。这个配置文件包含了连接到 MySQL 数据库所需的所有信息,以及其他与 Maxwell 相关的配置。

注意事项:

  • maxwell-bootstrap 命令通常只用于初始化数据同步,而不是用于持续的数据同步。一旦初始数据快照被创建并加载到 Maxwell 中,Maxwell 就会开始监听 binlogs 并进行实时数据同步。
  • 在执行 maxwell-bootstrap 命令之前,请确保 MySQL 数据库的 binlog 功能已经开启,并且 binlog 格式设置为 ROW 或 MIXED(以 ROW 为主)。这是因为 Maxwell 需要 binlog 中的行级更改信息来进行数据同步。
  • maxwell-bootstrap 命令的输出通常是一系列包含数据库更改事件的 JSON 消息。这些消息可以被直接发送到 Maxwell 的输出目标(如 Kafka),或者被写入到一个文件中供后续处理。
  • 由于 maxwell-bootstrap 会生成大量的数据,因此在执行该命令时要小心谨慎,确保目标系统能够处理这些数据负载。
  • 最后,请注意检查 Maxwell 的官方文档以获取关于 maxwell-bootstrap 命令的最新信息和最佳实践建议。因为随着 Maxwell 版本的更新,命令的功能和用法可能会发生变化。

2、查看kafka数据

{"database": "user","table": "t_user","type": "bootstrap-insert","ts": 1709735024,"data": {"id": 27,"name": "Hello 001","msg": "新用户注册事务提交事件","create_time": "2023-05-20 11:37:44","last_login_time": null,"status": 1}
}

示例输出:

{"database": "dbname","table": "tablename","type": "bootstrap-start","ts": 1234567890,"data": {}
}
{"database": "dbname","table": "tablename","type": "bootstrap-insert","ts": 1234567890,"data": {"id": 1,"name": "John Doe"}
}
{"database": "dbname","table": "tablename","type": "bootstrap-complete","ts": 1234567890,"data": {}
}

在这个示例中,type 字段用于区分不同类型的更改事件。bootstrap-startbootstrap-complete 分别表示数据快照的开始和结束,而 bootstrap-insert 则表示一个插入操作。ts 字段表示更改事件的时间戳,而 data 字段则包含了更改的具体数据。

参考

  • https://github.com/zendesk/maxwell

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

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

相关文章

LeNet5实战——衣服分类

搭建模型训练代码(数据处理、模型训练、性能指标)——> 产生权重w ——>模型结构c、w测试 配置环境 Pycharm刚配置的环境找不到了-CSDN博客 model.py 导入库 import torch from torch import nn from torchsummary import summary 模型搭…

河北省光伏展

光伏展是指光伏行业的展览会,也被称为太阳能展。光伏展一般是由光伏企业、科研机构、行业协会和专业展览公司等共同举办的。展会内容包括光伏产品、技术、设备、材料、应用等方面的展示和交流。 光伏展通常是光伏行业的重要盛事,吸引了全球范围内的光伏企…

npm镜像源地址

镜像源地址替换问题(重要) 2024 年 1 月 22 日 ,registry.npm.taobao.org 的 SSL 证书正式过期。 2022 年 5 月 淘宝源发布了公告: (大家应该没有太多关注哦,也包括我,哈哈) &am…

144.乐理基础-根三五音、大三和弦、小三和弦

内容参考于: 三分钟音乐社 上一个内容:143.乐理基础-和弦是什么?和声是什么?三和弦-CSDN博客 必须先看上一个内容,了解什么是和弦、什么是和声,以及三和弦的定义 上一个内容最后写了三和弦的定义&#x…

【C++ 学习】构造函数详解!!!

1. 类的6个默认成员函数的引入 ① 如果一个类中什么成员都没有,简称为空类。 ② 空类中真的什么都没有吗?并不是,任何类在什么都不写时,编译器会自动生成以下6个默认成员函数。 ③ 默认成员函数:用户没有显式实现&…

嵌入式学习第二十五天!(网络的概念、UDP编程)

网络: 可以用来:数据传输、数据共享 1. 网络协议模型: 1. OSI协议模型: 应用层实际收发的数据表示层发送的数据是否加密会话层是否建立会话连接传输层数据传输的方式(数据包,流式)网络层数据的…

基于YOLOv8深度学习的智能道路裂缝检测与分析系统【python源码+Pyqt5界面+数据集+训练代码】深度学习实战、目标检测、目标分割

《博主简介》 小伙伴们好,我是阿旭。专注于人工智能、AIGC、python、计算机视觉相关分享研究。 ✌更多学习资源,可关注公-仲-hao:【阿旭算法与机器学习】,共同学习交流~ 👍感谢小伙伴们点赞、关注! 《------往期经典推…

外包干了5天,技术退步明显。。。。。

在湖南的一个安静角落,我,一个普通的大专生,开始了我的软件测试之旅。四年的外包生涯,让我在舒适区里逐渐失去了锐气,技术停滞不前,仿佛被时间遗忘。然而,生活的转机总是在不经意间降临。 与女…

家庭关怀视角下对待患病亲人态度的重要性——评析母对病父大声呵斥的行为现象

在家庭生活中,尤其是面对家人身体不适或疾病困扰的时候,我们的态度和行为方式显得尤为重要。近期,社会上存在一种令人忧虑的现象,即某些家庭中,母亲因压力或其他原因对生病的父亲表现出不耐烦甚至大吼大叫的态度。这种…

警用移动执法远程视频监控方案:安防视频监控系统EasyCVR+4G/5G移动执法仪

一、背景需求 在现代城市管理中,移动执法仪视频监控方案正逐渐成为一种高效、便捷的管理工具。该方案通过结合移动执法仪和视频监控技术,实现了对城市管理现场的实时监控和取证,有效提升了城市管理水平和效率。 移动执法仪作为现场执法的重…

TypeScript 哲学 - Object Types

readonly 修饰对象和数组的 双向可分配性是不同的 只有有一个可选属性不是意味着必须 不能传空对象,:这个例子(两个属性可选)而是如果对象有额外属性,那么必须至少加一个 可选属性。只要你在传递的值和目标类型有一个…

大模型概念解析 | Prompt Engineering

注1:本文系"概念解析"系列之一,致力于简洁清晰地解释、辨析复杂而专业的概念。本次辨析的概念是:大模型中的Prompt Engineering 大模型概念解析 | Prompt Engineering 第一部分 通俗解释 在人工智能的世界里,有一群被称为大模型的巨无霸。它们就像是知识的海绵…

关于STM32G070RBTx单片机使用HAL库往flash写数据的过程中死机问题

1.单片机型号:STM32G070RBTx 2.出现的问题 根据库函数FLASH_If_Write()的使用,我们分析往flash写数据的过程是把uint8_t 类型的数据(p_data)以地址的形式强转成uint64类型的,在一包128字节的数据时一次存储8位,存16次(packet_size/8)&#x…

Java项目:基于SSM框架实现的二手车交易平台【源码+开题报告+任务书+毕业论文+答辩ppt】

一、项目简介 本项目是一套基于SSM框架实现的二手车交易平台 包含:项目源码、数据库脚本等,该项目附带全部源码可作为毕设使用。 项目都经过严格调试,eclipse或者idea 确保可以运行! 该系统功能完善、界面美观、操作简单、功能齐…

MySQL底层原理

1. 请解释MySQL的逻辑架构和物理架构。 MySQL的逻辑架构和物理架构涉及到多个层面,包括网络连接、服务处理、存储引擎以及数据存储等部分。具体如下: 逻辑架构: 连接层(Connection Layer):客户端通过TCP…

瑞芯微 | I2S-音频基础 -1

最近调试音频驱动,顺便整理学习了一下i2s、alsa相关知识,整理成了几篇文章,后续会陆续更新。 喜欢嵌入式、Li怒晓得老铁可以关注一口君账号。 1. 音频常用术语 名称含义ADC(Analog to Digit Conversion)模拟信号转换…

Android中Fragment的onResume方法的介绍、执行时机,以及不执行回调的异常情况分析

onResume()是Fragment生命周期中的一个重要方法,表示Fragment已经获取焦点并开始与用户交互。在onResume()方法中,Fragment通常完成与用户界面交互的准备工作,比如开始执行一些动画、加载数据或注册监听器等。 1. 回调时机: onRe…

stm32普通定时器脉冲计数(发送固定脉冲个数),控制步进电机驱动器

拨码开关设置驱动器,细分 方法思路:用通用定时器TIM2,1ms产生一次中断;在中断里做IO反转; 发送10个脉冲信号

系统架构设计师考试大纲

一、系统架构设计综合知识 1. 计算机系统基本知识 1.1 计算机系统概述 1.2 计算机硬件 1.2.1 计算机硬件组成 1.2.2 处理器 1.2.3 存储器 1.2.4 总线 1.2.5 接口 1.2.6 外部设备 1.3 计算机软件 1.3.1 计算机软件概述 1.3.2 操作系统 1.3.3 数据库系统 1.3.4 文件系统 1.3.5 网…

搬家微信小程序:便捷预约,轻松解决搬家难题

在快节奏的现代生活中,搬家成为许多人不得不面对的一项繁琐任务。从整理物品、联系搬家公司,到现场协调,每一个环节都让人倍感压力。然而,如今随着科技的不断发展,搬家微信小程序的出现,为这一难题带来了便…