flink-connector-mysql-cdc：01 mysql-cdc础配置代码演示

flink-connector-mysql-cdc：

  • 01 mysql-cdc基础配置代码演示
  • 02 mysql-cdc高级扩展
  • 03 mysql-cdc常见问题汇总
  • 04 mysql-cdc-kafka生产级代码分享
  • 05 flink-kafka-doris生产级代码分享
  • 06 flink-kafka-hudi生产级代码分享

flink-cdc版本：3.2.0

flink版本：flink-1.18.0

mysql版本：8.0.26

java版本：1.8

maven版本：3.8.4

目录

1. Mysql数据库设置

1.1 开启binlog日志

1.2 创建用户

1.3 准备测试数据 

2. 编写测试代码

2.1 maven 依赖

2.2 测试代码

3. mysql-cdc扩展

3.1 时区设置

3.2 为每个读取器设置不同的 SERVER ID

1. Mysql数据库设置

1.1 开启binlog日志

编辑 MySQL 配置文件

  • 在 Unix/Linux 系统中，通常是 /etc/my.cnf 或 /etc/mysql/my.cnf。

  • 在 Windows 上，可能位于 C:\ProgramData\MySQL\MySQL Server X.Y\my.ini。

# 在 mysqld 部分下添加以下内容（如果已经存在，请确认其值）：
[mysqld]
log-bin=mysql-bin  # 二进制日志文件前缀，MySQL将生成名为 mysql-bin.000001, mysql-bin.000002 等的文件。
binlog-format=row   # 设置二进制日志格式为行级（row），可选值为 STATEMENT、ROW 和 MIXED；这里推荐使用行级。
expire_logs_days=7  # 设置二进制日志的过期时间，单位为天，超过这个天数后的日志将被自动删除，这里以 7 天为例。
max-binlog-size=100M # 设置单个二进制日志文件的最大大小，超出后将自动创建一个新的日志文件（可以根据需要调整）。

1.2 创建用户

以 “flinkcdc”用户为例

# 创建 MySQL 用户：
CREATE USER 'flinkcdc'@'localhost' IDENTIFIED BY '123456';
# 向用户授予所需的权限：
GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'flinkcdc' IDENTIFIED BY '123456';
# 授权所有权限
GRANT ALL PRIVILEGES ON *.* TO 'flinkcdc'@'localhost';
GRANT ALL PRIVILEGES ON *.* TO 'flinkcdc'@'%';
# 完成用户的权限：
FLUSH PRIVILEGES;

1.3 准备测试数据 

# 使用flinkcdc用户登录数据库# 创建测试数据库
create database cdc_demo;
# 创建测试表
CREATE TABLE cdc_demo.flink_cdc_test (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(50) NOT NULL,description TEXT,age INT,balance DECIMAL(10, 2),is_active BOOLEAN DEFAULT TRUE,created_at DATETIME DEFAULT CURRENT_TIMESTAMP,birth_date DATE,long_value BIGINT,last_login TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
# 插入测试数据
INSERT INTO cdc_demo.flink_cdc_test (name, description, age, balance, is_active, created_at, birth_date, long_value, last_login) VALUES
('Alice Smith', 'Alice is a software engineer with 5 years of experience.', 30, 2500.50, TRUE, '2023-01-01 10:00:00', '1992-05-15', 12345678901234, '2023-05-20 10:00:00'),
('Bob Johnson', 'Bob enjoys hiking and outdoor activities.', 25, 1500.00, TRUE, '2023-02-15 12:30:00', '1998-08-22', 987654321054321, '2023-05-18 14:00:00'),
('Charlie Brown', 'Charlie is an avid reader and coffee lover.', 35, 3200.75, FALSE, '2023-03-22 14:45:00', '1988-01-11', 135792468012345, '2023-05-19 09:20:00'),
('Daisy Miller', 'Daisy loves painting and traveling.', 28, 1800.25, TRUE, '2023-04-05 09:15:00', '1994-11-03', 24681357901234, '2023-05-21 12:30:00'),
('Ethan White', 'Ethan enjoys playing guitar and writing songs.', 40, 5000.00, TRUE, '2023-05-18 16:20:00', '1983-07-30', 98765432102468, '2023-05-22 15:00:00');

2. 编写测试代码

2.1 maven 依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.toroidal</groupId><artifactId>flink-connector-mysql-cdc-demo</artifactId><name>flink-connector-mysql-cdc-demo</name><version>1.0-SNAPSHOT</version><repositories><repository><id>aliyunmaven</id><name>阿里云公共仓库</name><url>https://maven.aliyun.com/repository/public/</url></repository><repository><id>mirrorId</id><name>Human Readable Name for this Mirror.</name><url>http://my.repository.com/repo/path</url></repository></repositories><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><flink.version>1.18.0</flink.version><scala.binary.version>2.12</scala.binary.version><flinkcdc.version>3.2.0</flinkcdc.version><mysql.version>8.0.26</mysql.version><log4j.version>2.17.1</log4j.version><lombok.version>1.18.24</lombok.version><fastjson.version>1.2.83</fastjson.version></properties><dependencies><!-- flink Dependency --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-base</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-statebackend-rocksdb</artifactId><version>${flink.version}</version></dependency><!-- mysql-cdc --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>${flinkcdc.version}</version></dependency><!-- mysql --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>${mysql.version}</version></dependency><!-- json --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><!-- log --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId><version>${log4j.version}</version></dependency></dependencies><build><sourceDirectory>src/main/java</sourceDirectory><plugins><!-- 编译插件 --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.9.0</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.18.1</version><configuration><useFile>false</useFile><disableXmlReport>true</disableXmlReport><includes><include>**/*Test.*</include><include>**/*Suite.*</include></includes></configuration></plugin><!-- 打包插件(会包含所有依赖) --><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>2.3</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><filters><filter><artifact>*:*</artifact><excludes><!--zip -d learn_spark.jar META-INF/*.RSA META-INF/*.DSA META-INF/*.SF --><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><!-- 设置jar包的入口类(可选) --><mainClass>com.toroidal.mysql.MysqlCdcStreamApp</mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins></build>
</project>

2.2 测试代码

package com.toroidal.mysql;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;/*** @Author Toroidal* @Date 2024/12/04 14:42* @Version 1.0*/
public class MysqlCdcStreamApp {public static void main(String[] args) throws Exception {MySqlSource<String> mySqlSource = MySqlSource.<String>builder().hostname("localhost").port(3306).username("flinkcdc").password("123456")// 表所在的数据库库名，如果需要捕获整个库的表，配置为：".*".；如果需要捕获多个数据库配置为：.databaseList("cdc01", "cdc02").databaseList("cdc_demo")// 设置需要捕获日志的表名，注意需要配置库名，大小敏感.tableList("cdc_demo.flink_cdc_test")// 将 SourceRecord 转换为 JSON 字符串。.deserializer(new JsonDebeziumDeserializationSchema()).build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// enable checkpointenv.enableCheckpointing(3000);env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source").setParallelism(4).print().setParallelism(1);env.execute("Print MySQL Snapshot + Binlog");}
}

运行结果 

3. mysql-cdc扩展

3.1 时区设置

mysql-cdc读取出来的 timestamp 字段时区相差8小时，将时区和MySQL服务器时区设置一致即可：

查询当前数据库时区：

SELECT * FROM mysql.time_zone_name;

设置时区为东八时区

.serverTimeZone("Asia/Shanghai")

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder().hostname("localhost").port(3306).username("flinkcdc").password("123456")// 表所在的数据库库名，如果需要捕获整个库的表，配置为：".*".；如果需要捕获多个数据库配置为：.databaseList("cdc01", "cdc02").databaseList("cdc_demo")// 设置需要捕获日志的表名，注意需要配置库名，大小敏感.tableList("cdc_demo.flink_cdc_test").serverTimeZone("Asia/Shanghai")// 将 SourceRecord 转换为 JSON 字符串。.deserializer(new JsonDebeziumDeserializationSchema()).build();

3.2 为每个读取器设置不同的 SERVER ID

每个用于读取 binlog 的 MySQL 数据库客户端都应该有一个唯一的 ID，称为 server id。MySQL 服务器将使用此 id 来维护网络连接和 binlog 位置。因此，如果不同的作业共享相同的服务器 ID，则可能会导致从错误的 binlog 位置读取。 

.serverId("flink-cdc-01")
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder().hostname("localhost").port(3306)// 表所在的数据库库名，如果需要捕获整个库的表，配置为：".*".；如果需要捕获多个数据库配置为：.databaseList("cdc01", "cdc02").databaseList("cdc")// 设置需要捕获日志的表名，注意需要配置库名，大小敏感.tableList("cdc.flink_cdc_test").username("flinkcdc").serverTimeZone("Asia/Shanghai").serverId("flink-cdc-01").password("123456")// 将 SourceRecord 转换为 JSON 字符串。.deserializer(new JsonDebeziumDeserializationSchema()).build();

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

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

相关文章

ELK的Filebeat

ELK的Filebeat

目录 传送门前言一、概念1. 主要功能2. 架构3. 使用场景4. 模块5. 监控与管理 二、下载地址三、Linux下7.6.2版本安装filebeat.yml配置文件参考&#xff08;不要直接拷贝用&#xff09;多行匹配配置过滤配置最终配置&#xff08;一、多行匹配、直接读取日志文件、EFK方案&#…
阅读更多...
HTML 添加 文本水印

HTML 添加 文本水印

body,html {margin: 0;height: 100vh;width: 100vw;} // 自定义文案const setting {text: "水印文案", // 水印内容innerDate: true, // 在水印下方增加日期width: 110, // 水印宽度};// 自定义文字水印const watermark (function () {return {build: function (a…
阅读更多...
Android平台GB28181设备接入模块如何支持GB28181云端录像补录

Android平台GB28181设备接入模块如何支持GB28181云端录像补录

技术背景 GB28181 的补录功能是一种用于弥补视频数据缺失的重要机制。在实际的视频监控场景中&#xff0c;由于网络不稳定、设备故障等多种因素&#xff0c;可能会导致视频数据在上云或存储过程中出现缺失&#xff0c;无法保证数据的完整性。GB28181 的补录功能就是为了解决这…
阅读更多...
SpringBoot如何使用EasyExcel实现表格导出（简洁快速入门版本）

SpringBoot如何使用EasyExcel实现表格导出（简洁快速入门版本）

前言 前面给大家介绍了动态表头的导入&#xff0c;这篇文章给大家介绍如何实现导出 前面给大家介绍了动态表头的导入&#xff0c;我们了解了如何通过EasyExcel灵活地读取结构不固定的Excel文件。这次&#xff0c;我们将目光转向数据导出——即如何将数据以Excel文件的形式输出…
阅读更多...
Web 毕设篇-适合小白、初级入门练手的 Spring Boot Web 毕业设计项目：智行无忧停车场管理系统（前后端源码 + 数据库 sql 脚本）

Web 毕设篇-适合小白、初级入门练手的 Spring Boot Web 毕业设计项目：智行无忧停车场管理系统（前后端源码 + 数据库 sql 脚本）

&#x1f525;博客主页&#xff1a; 【小扳_-CSDN博客】 ❤感谢大家点赞&#x1f44d;收藏⭐评论✍ 文章目录 1.0 项目介绍 1.1 项目功能 2.0 用户登录功能 3.0 首页界面 4.0 车辆信息管理功能 5.0 停车位管理功能 6.0 入场登记管理功能 7.0 预约管理功能 8.0 收费规则功能 9.0…
阅读更多...
365天深度学习训练营-第P7周：马铃薯病害识别（VGG-16复现）

365天深度学习训练营-第P7周：马铃薯病害识别（VGG-16复现）

文为「365天深度学习训练营」内部文章 参考本文所写记录性文章&#xff0c;请在文章开头带上「&#x1f449;声明」 &#x1f37a; 要求&#xff1a; 自己搭建VGG-16网络框架【达成√】调用官方的VGG-16网络框架【达成√】如何查看模型的参数量以及相关指标【达成√】 &#…
阅读更多...
芯食代冻干科技研究院：创新与品质并重，推动家用冻干机高质量发展

芯食代冻干科技研究院：创新与品质并重，推动家用冻干机高质量发展

11月25日,芯食代首届食品冻干前沿与智能化升级创新大会在江苏常州成功举办。本次大会由芯食代冻干科技研究院(江苏)有限公司与芯食代(上海)科技发展有限公司联合主办,云集学界专家教授、商界企业精英,共议家用冻干机的未来创新发展。作为创新大会,芯食代冻干科技研究院也在本次…
阅读更多...
项目基于oshi库快速搭建一个cpu监控面板

项目基于oshi库快速搭建一个cpu监控面板

后端&#xff1a; <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.github.oshi</groupId><artifactId>oshi-…
阅读更多...
【数据结构】手搓链表

【数据结构】手搓链表

一、定义 typedef struct node_s {int _data;struct node_s *_next; } node_t;typedef struct list_s {node_t *_head;node_t *_tail; } list_t;节点结构体&#xff08;node_s&#xff09;&#xff1a; int _data;存储节点中的数据struct node_s *_next;&#xff1a;指向 node…
阅读更多...
毕设记录_音圈电机及电磁学相关_20241204

毕设记录_音圈电机及电磁学相关_20241204

前言 提醒&#xff1a; 文章内容为方便作者自己后日复习与查阅而进行的书写与发布&#xff0c;其中引用内容都会使用链接表明出处&#xff08;如有侵权问题&#xff0c;请及时联系&#xff09;。 其中内容多为一次书写&#xff0c;缺少检查与订正&#xff0c;如有问题或其他拓展…
阅读更多...
UPLOAD LABS | PASS 10 - 黑名单绕过（Windows . 绕过 - 变体）

UPLOAD LABS | PASS 10 - 黑名单绕过（Windows . 绕过 - 变体）

关注这个靶场的其它相关笔记&#xff1a;UPLOAD LABS —— 靶场笔记合集-CSDN博客 0x01&#xff1a;过关流程 本关的目标是上传一个 WebShell 到目标服务器上&#xff0c;并成功访问&#xff1a; 通过查看源码&#xff0c;可以发现&#xff0c;本关在之前所有关卡的基础上做了…
阅读更多...
【Elasticsearch】实现分布式系统日志高效追踪

【Elasticsearch】实现分布式系统日志高效追踪

&#x1f9d1; 博主简介&#xff1a;CSDN博客专家&#xff0c;历代文学网&#xff08;PC端可以访问&#xff1a;https://literature.sinhy.com/#/literature?__c1000&#xff0c;移动端可微信小程序搜索“历代文学”&#xff09;总架构师&#xff0c;15年工作经验&#xff0c;…
阅读更多...
FoldX(FoldX5)的安装流程

FoldX(FoldX5)的安装流程

下载地址:官网 https://foldxsuite.crg.eu/] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] ] zip解压yasaraPlugin.zip 我将foldx_20241231改为foldx vim ~/.bashrc 将foldx文件所在路径写进PATH vim ~/.bashrc我的…
阅读更多...
利用红黑树封装map，和set，实现主要功能

利用红黑树封装map，和set，实现主要功能

如果不知道红黑树是什么的时候可以去看看这个红黑树 思路 首先我们可以把封装分为两个层面理解&#xff0c;上层代码就是set,和map&#xff0c;底层就是红黑树 就相当于根据红黑树上面套了两个map,set的壳子&#xff0c;像下面这张图一样 对于map和set&#xff0c;map里面存…
阅读更多...
分类算法中的样本不平衡问题及其解决方案

分类算法中的样本不平衡问题及其解决方案

一、样本不平衡问题概述 在机器学习的分类任务中&#xff0c;样本不平衡是指不同类别训练样本数量存在显著差异的现象。这一差异会给模型训练和性能评估带来挑战&#xff0c;尤其在处理少数类样本时&#xff0c;模型可能难以有效学习其特征。 以二分类为例&#xff0c;理想情况…
阅读更多...
通过HTML Canvas 在图片上绘制文字

通过HTML Canvas 在图片上绘制文字

目录 前言 一、HTML Canvas 简介 二、准备工作 三、绘制图片 四、绘制文字 五、完整代码 效果演示&#xff1a; 前言 HTML canvas 为我们提供了无限的创意可能性。今天&#xff0c;我们就来探索一下如何通过 HTML canvas 将图片和文字绘制到图片上&#xff0c;创造出独特…
阅读更多...
MBox20边缘计算网关：氢能车间数据采集的智慧引擎

MBox20边缘计算网关：氢能车间数据采集的智慧引擎

氢能作为未来能源体系的重要组成部分&#xff0c;其安全、高效、环保的特性备受瞩目。在氢能车间的日常运营中&#xff0c;数据采集是确保生产流程优化、设备稳定运行及能效提升的关键环节。然而&#xff0c;面对氢能车间复杂多变的生产环境和海量数据&#xff0c;如何实现高效…
阅读更多...
linux环境GitLab服务部署安装及使用

linux环境GitLab服务部署安装及使用

一、GitLab介绍 GitLab是利用Ruby onRails一个开源的版本管理系统&#xff0c;实现一个自托管的Git项目仓库&#xff0c;可通过Web界面进行访问公开的或者私人项目。 二、GitLab安装 1、先安装相关依赖 yum -y install policycoreutils openssh-server openssh-clients postf…
阅读更多...
Gartner报告解读（四）| 如何运用上升期的基础设施自动化（IA）为企业数字化转型赋能？

Gartner报告解读（四）| 如何运用上升期的基础设施自动化（IA）为企业数字化转型赋能？

近期&#xff0c;Gartner发布的《2024年中国基础设施战略技术成熟度曲线》显示&#xff0c;未来5-10年&#xff0c;大量具有颠覆性或较高影响力的创新技术可能会实现主流采用&#xff0c;其中就包括基础设施自动化&#xff08;IA&#xff09;。 基础设施自动化Gartner评估情况 …
阅读更多...
请求响应：常见参数接收及封装（Json参数及路径参数）

请求响应：常见参数接收及封装（Json参数及路径参数）

Json参数 Json格式的数据具有轻量级、易于阅读和编写、易于解析等诸多优点。在前后端交互时&#xff0c;大部分情况下请求体中的数据会以JSON格式进行传递。前端的请求在请求体中携带了Json格式数据&#xff0c;后端程序需要对其进行解析并封装使用&#xff0c;而接收Json参数…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023