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

news/2025/2/5 22:45:06/

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 测试代码

mysql-cdc%E6%89%A9%E5%B1%95-toc" style="margin-left:0px;">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");}
}

运行结果 

mysql-cdc%E6%89%A9%E5%B1%95">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.ppmy.cn/news/1552609.html

相关文章

实战ansible-playbook(四) -文件操作重定向/追加

实战ansible-playbook(四) -文件操作重定向/追加

原始命令： ----------阶段1--------------- apt-get update -y apt install nano vim iputils-ping net-tools dialog gcc apt-utils make -y systemctl stop unattended-upgradessystemctl disable unattended-upgradesecho APT::Periodic::Update-Package-Lists "1&qu…
阅读更多...
59 基于STM32的烟雾、红外、温湿度检测

59 基于STM32的烟雾、红外、温湿度检测

所有仿真详情导航： PROTEUS专栏说明-CSDN博客 目录 一、主要功能 二、硬件资源 三、主程序编程 四、资源下载 一、主要功能 基于SMT32F103C8T6单片机，采用DHT11检测温湿度，采用光敏电阻检测光照，采用滑动变阻器分别模拟红外、烟雾，通过OLED显示屏显示，如果湿度过低…
阅读更多...
WEB安全 PHP学习

WEB安全 PHP学习

PHP基础 PHP编码显示问题 header ("Content-type: text/html; charsetgb2312"); header("Content-Type: text/html;charsetutf-8"); windows需要使用gbk编码显示 源码是 <?php header ("Content-type: text/html; charsetgb2312"); sys…
阅读更多...
Python几种常用数据结构（重制版）

Python几种常用数据结构（重制版）

一、列表 [List] 定义&#xff1a;有序可重复的数据集合。示例&#xff1a;my_list [element1, element2, element3]增加元素方法&#xff1a; append()&#xff1a;在列表末尾增加单个元素&#xff08;列表特有方法&#xff09;&#xff0c;例如 my_list.append(element)。e…
阅读更多...
HOG 算法变形：原理、应用与创新发展

HOG 算法变形：原理、应用与创新发展

摘要&#xff1a; 方向梯度直方图&#xff08;Histogram of Oriented Gradients&#xff0c;HOG&#xff09;算法在计算机视觉领域有着广泛的应用。本文深入探讨了 HOG 算法的原理&#xff0c;详细阐述了其在不同方面的变形&#xff0c;包括特征提取方式的改进、分块策略的优化…
阅读更多...
前端学习笔记-Vue篇-01

前端学习笔记-Vue篇-01

第一章 Vue核心 1.1 Vue的基本认识 1.1.1 官网 英文官网Vue.js - The Progressive JavaScript Framework | Vue.js 中文官网Vue.js - 渐进式 JavaScript 框架 | Vue.js 【注意&#xff1a;Vue2&#xff1a;Vue.js | 介绍 — Vue.js】 1.1.2 介绍描述 渐进式JavaScript框…
阅读更多...
Asp.net Mvc在VSCore中如何将增删改查的增改添加数据传输到页面(需配合上一篇Mvc的增删改查一起）

Asp.net Mvc在VSCore中如何将增删改查的增改添加数据传输到页面(需配合上一篇Mvc的增删改查一起）

Linq集成查询&#xff08;关联Lambda&#xff09; First FirstOrDefault 找到第一个符合条件的元素 First(x >x.Id id) 返回第一个Id等于id的元素&#xff0c;如果都没有符合的&#xff0c;报错FirstOrDefault(x >x.Id id) 返回第一个Id等于id的元素&#xff0c;如果…
阅读更多...
UML箭线图的理解和实践

UML箭线图的理解和实践

在软件开发的世界里&#xff0c;UML&#xff08;统一建模语言&#xff09;作为一种标准化的建模语言&#xff0c;扮演着举足轻重的角色。UML类图更是软件开发设计和架构过程中的核心工具&#xff0c;它不仅能帮助开发者明确系统中的类及其关系&#xff0c;还能为后续的代码实现…
阅读更多...
最新文章

Copyright @ 2022~2023