在当今数据快速增长的时代，实时、高效地同步和处理来自各种数据源的信息成为了企业和开发者面临的重要挑战。

MySQL作为广泛使用的数据库之一，其变更数据捕获（CDC）功能对于实现这一目标至关重要。在这篇文章中，我们将深入探讨MySQL CDC源连接器在SeaTunnel框架下的应用，涵盖从基础设置到高级配置的各个方面。

MySQL CDC源连接器

支持的引擎

SeaTunnel Zeta<br/> Flink <br/>

主要特性

• [ ] 批量
• [x] 流式
• [x] 精确一次
• [ ] 列投影
• [x] 并行处理
• [x] 支持用户定义的拆分

描述

MySQL CDC连接器允许从MySQL数据库读取快照数据和增量数据。本文档描述了如何设置MySQL CDC连接器以对MySQL数据库运行SQL查询。

支持的数据源信息

数据库依赖

安装Jdbc驱动程序

请将mysql驱动程序下载并放入${SEATUNNEL_HOME}/lib/目录中。例如：cp mysql-connector-java-xxx.jar $SEATNUNNEL_HOME/lib/

创建MySQL用户

您必须为Debezium MySQL连接器监视的所有数据库定义一个具有适当权限的MySQL用户。

1. 创建MySQL用户：

mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';

2. 为用户授予所需权限：

mysql> GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';

3. 完成用户的权限设置：

mysql> FLUSH PRIVILEGES;

启用MySQL binlog

为了实现MySQL的复制，必须启用二进制日志。二进制日志记录了用于复制工具传播更改的事务更新。

1. 检查log-bin选项是否已经开启：

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| binlog_row_image         | FULL           |
| enforce_gtid_consistency | ON             |
| gtid_mode                | ON             |
| log_bin                  | ON             |
+--------------------------+----------------+
5 rows in set (0.00 sec)

2. 如果与上述结果不一致，请使用以下属性配置您的MySQL服务器配置文件（$MYSQL_HOME/mysql.cnf），如下表所示：

# 启用二进制复制日志并设置前缀、过期时间和日志格式。
# 前缀是任意的，过期时间对于集成测试可以短一些，但在生产系统中会更长。
# 行级信息对于摄取工作是必需的。
# 服务器ID在生产系统上是必需的，但会有所不同。
server-id         = 223344
log_bin           = mysql-bin
expire_logs_days  = 10
binlog_format     = row
binlog_row_image  = FULL

# 启用gtid模式
gtid_mode = on
enforce_gtid_consistency = on

3. 重启MySQL服务器

/etc/inint.d/mysqld restart

4. 再次确认您的更改，通过再次检查binlog状态：

mysql> show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');
+--------------------------+----------------+
| Variable_name            | Value          |
+--------------------------+----------------+
| binlog_format            | ROW            |
| binlog_row_image         | FULL           |
| enforce_gtid_consistency | ON             |
| gtid_mode                | ON             |
| log_bin                  | ON             |
+--------------------------+----------------+
5 rows in set (0.00 sec)

注意

设置MySQL会话超时

当为大型数据库创建初始一致快照时，在读取表期间，已建立的连接可能会超时。您可以通过在MySQL配置文件中配置interactive_timeout和wait_timeout来防止这种行为。

• interactive_timeout：服务器等待交互连接活动关闭之前的秒数。有关更多详细信息，请参阅MySQL文档。
• wait_timeout：服务器等待非交互连接活动关闭之前的秒数。有关更多详细信息，请参阅MySQL文档。

有关更多数据库设置，请参见Debezium MySQL连接器

数据类型映射

源选项

Task Example

Simple

Support multi-table reading

env {
  parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 10000
}

source {
  MySQL-CDC {
    catalog = {
      factory = MySQL
    }
    base-url = "jdbc:mysql://localhost:3306/testdb"
    username = "root"
    password = "root@123"
    table-names = ["testdb.table1", "testdb.table2"]
    
    startup.mode = "initial"
  }
}

sink {
  Console {
  }
}

Support debezium-compatible format send to kafka

Must be used with kafka connector sink, see compatible debezium format for details

Changelog

• Add MySQL CDC Source Connector

next version

通过对MySQL CDC源连接器的深入了解，我们不仅能够更好地掌握数据同步的核心机制，还能有效提升数据处理的效率和精度。

无论是在数据集成、实时分析还是其他复杂的数据处理场景中，MySQL CDC源连接器都将成为SeaTunnel用户强大的助手。随着数据技术的不断进步，期待看到更多创新和优化在未来版本中的实现，为开发者带来更多便利和可能。

本文由 白鲸开源科技 提供发布支持！