MySQL 迁移同步到 Kafka
NineData 数据复制支持将 MySQL 中的数据复制到 Kafka,实现数据的实时流转和分发。
背景信息
Kafka 是一个分布式消息队列,具有高吞吐量、低延迟、高可靠性和可扩展性等特点,适用于实时数据处理和分发的场景。将 MySQL 中的数据复制到 Kafka 可以构建实时数据流处理系统,满足实时数据处理、数据分析、数据挖掘、业务监控等需求。此外,将 MySQL 中的数据复制到 Kafka 中还可以实现数据的解耦,从而减少系统间的依赖和耦合,使系统更易于维护扩展。
投递格式可以选择 Canal JSON、Debezium JSON 或 Avro。Canal JSON 的字段说明请参见附录 1,Debezium JSON 的消息结构请参见附录 2,Avro Schema 的结构和编码请参见附录 3。使用 Avro 时,可以通过 Confluent Schema Registry 管理消息 Schema。
前提条件
已将源和目标数据源添加至 NineData。如何添加,请参见创建数据源。
支持的 MySQL 数据源版本为 5.1 ~ 9.0。
目标的数据库类型为 Kafka 0.10 及以上版本。
源数据源必须开启 Binlog,并且 Binlog 相关参数设置如下:
binlog_format=ROWbinlog_row_image=FULL
提示如果源数据源为备库,为保证获取完整的 Binlog 日志,还需要开启
log_slave_updates参数。
使用限制
- 数据复制功能仅针对数据源中的用户数据库,系统库不会被复制。例如:MySQL 类型数据源中的
information_schema、mysql、performance_schema、sys库不会被复制。 - 执行数据同步前需评估源数据源和目标数据源的性能,同时建议业务低峰期执行数据同步。否则全量数据初始化时将占用源数据源和目标数据源一定的读写资源,导致数据库负载上升。
- 需要确保同步对象中的每张表都有主键或唯一约束、列名具有唯一性,否则可能会重复同步相同数据。
操作步骤
登录 NineData 控制台。
在左侧导航栏单击数据复制 > 数据复制。
在数据复制页面,单击右上角的创建复制。
在数据源与目标页签,按照下表进行配置,并单击下一步。
参数 说明 任务名称 输入数据同步任务的名称,为了方便后续查找和管理,请尽量使用有意义的名称。最多支持 64 个字符。 源数据源 同步对象所在的数据源。 目标数据源 接收同步对象的数据源。 Kafka Topic 选择目标 Kafka Topic(主题),源数据源中的数据将写入到该指定 Topic 中。 投递分区 将数据投递到一个 Topic 中时,可以指定这些数据要投递到 Topic 中的哪个分区。 - 全部投递到 Partition 0:将所有数据投递到默认的分区 0。
- 按【库名 + 表名】的 Hash 值散列投递到不同 Partition:将数据散列分配到不同的分区中,系统会使用库名和表名的哈希值来计算目标数据投递到哪个分区,以确保散列投递过程中,同一个表中的数据被投递到同一分区。
投递数据格式 选择投递到 Kafka 的消息格式。 - Canal JSON:以 Canal JSON 格式投递消息。
- Avro:以 Avro 格式投递消息,并通过 Confluent Schema Registry 管理消息 Schema。
Confluent Schema Registry服务地址 选择 Avro 时必填。输入 Confluent Schema Registry 服务地址,例如 <schema-registry-host>:<port>。用户名或API-KEY 选择 Avro 时按需填写 Schema Registry 用户名或 API Key。 密码或Secret 选择 Avro 时按需填写 Schema Registry 密码或 Secret。 目标对象命名规则 选择对象名称从源端迁移到目标端后的大小写转换规则。 - 全部转小写:无论源端的命名规则如何,目标端的命名规则全部为小写。
- 保持与源一致:沿用源端的命名规则。
- 全部转大写:无论源端的命名规则如何,目标端的命名规则全部为大写。
复制类型 选择复制类型。 - 全量复制:同步源数据源的所有对象和数据,即全量数据复制。右侧的开关为周期性全量复制的开关,更多信息,请参见周期性全量复制。
- 增量复制:在全量同步完成后,基于源数据源的日志进行增量同步。
增量开始时间 复制类型仅为增量复制时需要选择。 - 从启动时间开始:以当前复制任务开始时间为基准,进行增量复制。
- 自定义时间:选择增量复制开始的时间点,您可以根据您的业务所在地域按需选择时区。如果将时间点配置为当前复制任务开始前,则如果该时间段内有 DDL 操作,复制任务将失败。
复制规格 复制任务的规格,规格越大复制的速率越高,将鼠标悬浮于 图标上即可查看每个规格对应的速率和配置信息。如果先配置数据复制任务再购买资源,可以在此选择需要的规格;如果先购买资源再配置任务,系统会选中购买资源时指定的规格,且无法在任务配置中变更。
目标表存量数据处理策略(选中全量复制时需要选择) - 预检查报错并停止任务:预检查阶段检测到目标表中存在数据时,停止任务。
- 忽略目标存量数据,追加写入:预检查阶段检测到目标表中存在数据时,忽略该部分数据,追加写入其他数据。
- 清空目标存量数据,重新写入:预检查阶段检测到目标表中存在数据时,删除该部分数据,重新写入。
目标表增量数据冲突处理策略(选中增量复制时需要选择) - 运行时报错:增量复制过程中遇到目标数据已存在则报错,等待人工介入处理。
- 不更新目标数据:增量复制过程中遇到目标数据已存在则不写入,并继续后续任务。
- 更新目标数据:增量复制过程中遇到目标数据已存在则覆盖写入目标数据。
在选择复制对象页签,配置下列参数,然后单击下一步。
参数 说明 复制对象 选择需要复制的内容,您可以选择全部实例复制源库所有内容,也可以选择自定义对象,在源对象列表中选中需要复制的内容,单击>添加到右侧目标对象列表。 如果您需要创建多条相同复制对象的复制链路,可以创建一个配置文件,在新建任务的时候导入即可。单击右上角的导入配置,再单击下载模板,将配置文件模版下载到本地,编辑完成后单击上传文件上传该配置文件即可实现批量导入。字段和示例请参见导入复制对象配置模板。配置文件说明:
参数 说明 示例 source_table_name需同步的源表名。填写源数据源中实际存在的表名。 example_tbl_nametarget_table_name接收同步对象的目标表名。 example_tbl_namesource_database_name需同步的对象所在的源库名。 test_dbtarget_database_name接收同步对象的目标库名。 test_db_bakcolumn_list需要同步的字段列表。 ["col1", "clo2", "col3"]extra_configuration额外配置的 JSON 字符串。 column_rules用于配置字段映射与取值规则,其中column_name为源列名,target_column_name为目标列名,target_column_type为目标列类型,column_value为字段值;filter_condition用于配置行级过滤条件。{"extra_config": {"column_rules": [...]}}提示以下 JSON 表示下载的 Excel 模板中的两行示例数据。实际上传时,请保留 Excel 模板的列名,并将
extra_configuration单元格填写为 JSON 字符串;以下代码块不是可直接上传的 JSON 文件。[
{
"source_table_name": "example_tbl_name",
"target_table_name": "example_tbl_name",
"source_database_name": "test_db",
"target_database_name": "test_db_bak",
"column_list": ["col1", "clo2", "col3"],
"extra_configuration": {
"extra_config": {
"column_rules": [
{
"column_name": "example_column_name",
"target_column_name": "example_target_column_name",
"target_column_type": "int",
"column_value": "current_timestamp()"
},
{
"column_name": "example_column_name",
"target_column_name": "example_target_column_name",
"target_column_type": "int"
}
],
"filter_condition": "example_condition_expression"
}
}
},
{
"source_table_name": "example_order_tbl",
"target_table_name": "example_order_tbl",
"source_database_name": "test_db",
"target_database_name": "test_db_bak",
"column_list": ["col1", "clo2", "col3"],
"extra_configuration": {
"extra_config": {
"column_rules": [
{
"column_name": "example_column_name",
"target_column_name": "example_target_column_name",
"target_column_type": "int",
"column_value": "current_timestamp()"
},
{
"column_name": "example_column_name",
"target_column_name": "example_target_column_name",
"target_column_type": "int"
}
],
"filter_condition": "example_condition_expression"
}
}
}
]在配置映射页签,可以单独配置需要复制到 Kafka 的每个列,默认情况下将复制所选表的全部列。如果在配置映射阶段,源和目标数据源中有更新,可以单击页面右上角的刷新元数据按钮,重新获取源和目标数据源的信息。配置完成后,单击保存并预检查。
在预检查页签等待系统完成预检查。预检查通过后,在复制任务列表中打开已保存任务的详情页并启动任务。
您可以勾选开启数据一致性对比。在同步任务完成后,自动开启基于源数据源的数据一致性对比,保证两端数据一致。根据您选择的复制类型,开启数据一致性对比的启动时机如下:
- 全量复制:全量复制完成后启动。
- 全量复制+增量复制、增量复制:当增量数据首次和源数据源一致且延迟为 0 秒时启动。您可以单击查看详情,在复制详情页面中查看同步延迟。

如果预检查未通过,需要单击目标检查项右侧操作列的详情,排查失败的原因,手动修复后单击重新检查重新执行预检查,直到通过。
检查结果为警告的检查项,可视具体情况修复或忽略。
复制任务启动后,您可以进行以下操作:
单击查看详情查看同步任务各个阶段的执行情况。
- 单击返回列表可以返回数据复制任务列表页面。
修改增量同步位点
对于复制类型中包含增量复制且任务详情提供位点管理的任务,可以查看并修改同步位点。可用的位点类型和输入格式由当前数据源及复制链路决定,以页面显示为准;如果位点显示为 - 或位点类型没有可选项,则当前任务没有可供修改的位点。
该功能用于特定运维场景:重新指定增量任务的日志读取位置或同步写入位置。读取位点控制任务从源端哪个位置继续读取,写入位点控制任务从哪个位置继续处理同步写入;提交后任务会暂停并重新启动,使新的位点生效。
在复制任务列表中,单击目标任务的任务 ID,进入任务详情。
单击增量复制页签,然后单击同步位点管理,查看当前的读取位点和写入位点。
单击修改位点。在位点类型中选择要调整的读取位点或写入位点,再根据页面显示的位点类型填写调整位点。
确认值和影响范围后,单击验证并提交。任务会进入暂停和重新启动过程;任务恢复运行后,再次打开同步位点管理确认位点已更新。
修改同步位点,将会调整日志读取或同步写入的点位,可能导致数据丢失,谨慎操作! 修改任一位点都可能导致数据丢失或源端与目标端数据不一致。请仅在已授权的测试环境,或已完成备份并具备回滚方案的维护窗口执行。已完成增量复制的任务不能重置位点。
Kafka 投递设置补充
当一个 Kafka Topic 供任务中的复制对象共用时,投递分区设置提供按表名和主键值的 Hash 值散列到不同 Partition 的策略,对应选项为按【表名+主键值】的Hash值散列投递到不同 Partition。选择每张表使用独立 Topic 时,页面不显示投递分区设置。
创建任务时,在 Kafka 目标配置中选择 Topic 的分配方式:为任务中的复制对象共用一个 Topic 时,通过Kafka Topic选择目标 Topic;为每张表使用独立 Topic 时,无需选择共用 Topic,系统会为每张表生成一个 Topic。后者不显示 Topic 选择器和投递分区设置。
按表生成的 Topic 名称由 tableTopicPrefix、源库名和源表名依次用点号连接。例如,前缀为 ninedata 时,名称格式为 ninedata.<databaseName>.<tableName>。
按主键散列时,同一主键的 INSERT、UPDATE 和 DELETE 消息保持顺序,不同主键的行可以分散到不同 Partition 并行消费。没有主键的表回退为按库名和表名散列。该策略会逐条消息计算分区,因此批次中的多行会拆分为单行消息。Hash 策略只保证表或主键粒度的顺序,不保证跨表或跨主键的全局顺序;需要全局顺序时,选择全部投递到 Partition 0。
需要投递 Debezium JSON 消息时,在投递数据格式设置中选择 Debezium JSON;消息结构请参见附录 2。Avro Schema 结构和编码说明请参见附录 3。
查看同步结果
登录 NineData 控制台。
在左侧导航栏单击数据复制 > 数据复制。
在数据复制页面单击目标同步任务的 ID,打开复制详情页面,页面说明如下。

序号 功能 说明 1 同步延迟 源数据源和目标数据源之间的数据同步延迟,0 秒表示两端之间没有延迟,即 Kafka 端的数据目前已追平源端。 2 配置告警 配置告警后,系统会在任务失败时通过您选择的方式通知您。更多信息,请参见运维监控简介。 3 更多 - 暂停:暂停任务,仅状态为运行中的任务可选。
- 终止:结束未完成或监听中(即增量同步中)的任务,终止任务后无法重启任务,请谨慎操作。
- 删除:删除任务,任务删除后无法恢复,请谨慎操作。
4 全量复制(包含全量复制的场景下显示) 展示全量复制的进度和详细信息。 - 单击页面右侧的监控:查看全量复制过程中的各监控指标。全量复制过程中,还可以单击监控指标页面右侧的限流设置,限制每秒写入到目标数据源的速率。单位为 MB/S。
- 单击页面右侧的日志:查看全量复制的执行日志。
- 单击页面右侧的
图标:查看最新的信息。
5 增量复制(包含增量复制的场景下显示) 展示增量复制的各项监控指标。 - 单击页面右侧的限流设置:限制每秒写入到目标数据源的速率。单位为行/秒。
- 单击页面右侧的日志:查看增量复制的执行日志。
- 单击页面右侧的
图标:查看最新的信息。
6 修改对象 展示同步对象的修改记录。 - 单击页面右侧的修改同步对象,可对同步对象进行配置。
- 单击页面右侧的
图标:查看最新的信息。
7 展开 展示当前复制任务的详细信息,包括复制类型、复制对象、开始时间等。
附录 1:Canal JSON 数据格式说明
MySQL 迁移到 Kafka 的数据将以 JSON 格式存储,系统会均等划分 MySQL 中的数据存到多个 JSON 对象中,每个 JSON 代表一个消息(Message)。
- 全量复制阶段:单个消息存储的 MySQL 数据行数由 message.max.bytesmessage.max.bytes 是 Kafka 集群中允许的最大消息大小,默认值:
1000000字节(即 1MB)。您可以通过修改 Kafka 配置文件中的message.max.bytes参数来调整该值,这样每个消息中将会存储更多的 MySQL 数据行数。请注意,由于 Kafka 需要为每个消息分配内存空间,因此该值如果设置得过大,可能导致 Kafka 集群性能下降。 参数决定。 - 增量复制阶段:单个消息存储一行 MySQL 数据。
| 字段 | 字段类型 | 字段描述 | 字段示例 |
|---|---|---|---|
| serverId | STRING | 消息归属的数据源信息,格式:<连接地址:端口>。 | "serverId":"47.98.224.21:3307" |
| id | LONG | 消息的 Record id。本字段全局递增,用作消息重复消费的判断依据。 | "Id":156 |
| es | INT | 不同的任务阶段代表不同的含义:
| "es":1668650385 |
| ts | INT | 更多数据被投递到 Kafka 的时间,以 Unix 时间戳表示。 | "ts":1668651053 |
| isDdl | BOOLEAN | 数据是否为 DDL,取值:
| "is_ddl":true |
| type | STRING | 数据的类型,取值:
| "type":"INIT" |
| database | STRING | 数据所属的数据库。 | "database":"database_name" |
| table | STRING | 数据所属的表。如果 DDL 语句对应的对象为非表,则该字段取值为 null。 | "table":"table_name" |
| mysqlType | JSON | 记录数据在 MySQL 中的数据类型,以 JSON 数组表示。 | "mysqlType": {"id": "bigint(20)", "shipping_type": "varchar(50)" } |
| sqlType | JSON | 预留字段,无需关注。 | "sqlType":null |
| pkNames | ARRAY | Binlog 中的记录(Record)所对应的主键(Primary Key)名称。取值:
| "pkNames": ["id", "uid"] |
| data | ARRAY[JSON] | MySQL 投递到 Kafka 的数据,存储在 JSON 格式的数组中。
| "old": [{ "name": "someone", "phone": "(737)1234787", "email": "someone@example.com", "address": "somewhere", "country": "china" }] |
| old | ARRAY[JSON] | 记录 MySQL 到 Kafka 的增量复制详情。
除此之外的其他操作,该字段的值均为 null。 | "old": [{ "name": "someone", "phone": "(737)1234787", "email": "someone@example.com", "address": "somewhere", "country": "china" }] |
| sql | STRING | 如果当前数据是增量的 DDL 操作,则记录该操作对应的 SQL 语句。除此之外的其他操作,该字段的值均为 null。 | "sql":"create table sbtest1(id int primary key,name varchar(20))" |
附录 2:Debezium JSON 数据格式说明
每条 Kafka 消息是一个 Debezium JSON Envelope,顶层包含 schema 和 payload。schema 使用 Kafka Connect 风格描述 payload 的字段结构;同名且结构相同的表会复用同一份 Schema。DML 批次会拆分为单行消息,每条消息描述一行数据。
| 字段 | 类型 | 说明 |
|---|---|---|
schema | JSON | 消息 Schema,描述 payload 中各字段的类型结构。 |
payload | JSON | 消息数据,包含行镜像、事件来源信息和操作类型。 |
payload 中的主要字段如下:
| 字段 | 类型 | 说明 |
|---|---|---|
before | JSON | 变更前的行数据。全量快照和 INSERT 时为 null;UPDATE 时为更新前的整行;DELETE 时为被删除的整行。 |
after | JSON | 变更后的行数据。全量快照、INSERT 和 UPDATE 时为变更后的整行;DELETE 时为 null。 |
source | JSON | 事件来源元数据,包含连接器、库表和 Binlog 位点等信息。 |
op | STRING | 操作类型。r 表示全量快照读取,c 表示 INSERT(包括 REPLACE),u 表示 UPDATE,d 表示 DELETE,m 表示 DDL。 |
ts_ms | LONG | 消息投递到 Kafka 的时间戳,单位为毫秒。 |
databaseName、schemaName、database、schema、table、ddl | STRING | 仅 DDL 消息携带。ddl 为原始 SQL;database、schema 和 table 是兼容字段。DDL 消息不携带 before 和 after。 |
payload.source 中,version 为消息格式版本 ninedata-json-2,MySQL 的 connector 值为 mysql;snapshot 为字符串,"true" 表示全量快照,"false" 表示增量事件。payload.source.ts_ms 表示源事件时间戳,payload.ts_ms 表示消息投递到 Kafka 的时间戳,单位均为毫秒。全量快照消息的 op 为 r。
以下为 UPDATE 消息的 payload 节选,Schema、来源元数据和时间字段已省略:
{
"before": { "id": 20, "status": "PENDING" },
"after": { "id": 20, "status": "ACTIVE" },
"op": "u"
}
附录 3:Avro Schema 定义
Avro 消息使用 Confluent Schema Registry 管理 Schema。在一表一个 Topic 模式下,value subject 为 {topicName}-value。Kafka 消息的 value 按顺序包含:固定标记 0x00(1 字节)、Schema Registry 中对应 Schema 的编号(Schema ID,占 4 字节,按高位字节在前的顺序存储),以及使用 Avro Binary 编码的消息数据。消费者先读取 Schema ID,从 Schema Registry 获取对应的 Schema,再用该 Schema 解码后续消息数据。
消息 Schema 的顶层记录为 CdcMessage,包含变更元数据和根据源表结构动态生成的行记录。顶层字段如下:
| 字段 | Avro 类型 | 说明 |
|---|---|---|
serverId | null 或 string | 源端服务器标识。 |
id | null 或 long | 消息记录 ID。 |
timestamp | long,逻辑类型 timestamp-millis | 源 Binlog 事件时间,单位为毫秒。 |
sinkTimestamp | long | 消息投递到 Kafka 的时间,单位为毫秒。 |
changeSequence | null 或 string | 变更序列标识。 |
operation | CdcOperation 枚举 | 变更操作类型,取值见下表。 |
externalSchemaId | int | 消息 Schema 在 Schema Registry 中的 ID。 |
isDdl | boolean | 是否为 DDL 消息。 |
database | string | 源数据库名。 |
schema | null 或 string | 源 schema;MySQL 场景通常为 null。 |
table | null 或 string | 源表名;不对应表对象的 DDL 消息中为 null。 |
pkNames | null 或 array<string> | 主键列名列表;DDL 消息中为 null。 |
sqlType | null 或 map<string, string> | 列名到 MySQL 原始列类型的映射;DDL 消息中为 null。 |
data | null 或动态行 Record | 变更后的行数据,具体取值见下表。 |
beforeData | null 或动态行 Record | 变更前的行数据,具体取值见下表。 |
sql | null 或 string | 原始 DDL 语句;非 DDL 消息中为 null。 |
operation 的枚举值及编码序号如下:
| 枚举值 | 编码序号 | 含义 |
|---|---|---|
INIT | 0 | 全量初始化数据。 |
INSERT | 1 | INSERT Binlog 事件,包括 REPLACE。 |
UPDATE | 2 | UPDATE Binlog 事件。 |
DELETE | 3 | DELETE Binlog 事件。 |
DDL | 4 | DDL 语句。 |
data、beforeData 和 sql 会根据操作类型填写,具体如下:
| 操作 | data | beforeData | sql |
|---|---|---|---|
INIT | 初始化行 | null | null |
INSERT(包括 REPLACE) | 插入后的行 | null | null |
UPDATE | 更新后的行 | 更新前的行 | null |
DELETE | null | 删除前的行 | null |
DDL | null | null | 原始 DDL 语句 |
行记录按表结构动态生成,记录名由表名转换为 Avro 合法名称后加 _DATA,命名空间为 com.jzcloud.kafka.avro。列字段使用可空联合类型 ['null', <列类型>];列名转换为 Avro 合法名称时,原列名保存在 jzcloud.column.name 属性中。列的 Avro 类型取决于 MySQL 列类型和定义;sqlType 字段记录各列对应的 MySQL 原始类型。
以下示例展示一个 VARCHAR 列对应的动态行记录。实际字段由源表结构决定;此示例不是完整的 CdcMessage Schema。消费者应根据消息中的 Schema ID 从 Schema Registry 获取实际 Schema:
{
"type": "record",
"name": "example_DATA",
"namespace": "com.jzcloud.kafka.avro",
"fields": [
{
"name": "status",
"type": ["null", "string"]
}
]
}
使用 Avro 前,需要配置 Confluent Schema Registry 地址;Registry 开启认证时还需配置用户名和密码。完整的 MySQL 列类型映射取决于源列定义。
附录 4:预检查项一览表
| 检查项 | 检查内容 |
|---|---|
| 源数据源连接检查 | 检查源数据源网关状态、实例是否可达、用户名及密码准确性 |
| 目标数据源连接检查 | 检查目标数据源网关状态、实例是否可达、用户名及密码准确性 |
| 源库权限检查 | 检查源数据库的账号权限是否满足要求 |
| 源数据库 log_slave_updates 是否支持 | 当源数据库为从库时,检查 log_slave_updates 是否为 ON |
| 源数据源和目标数据源版本检查 | 检测源数据库与目标数据库的版本是否兼容 |
| 源数据库是否开启 Binlog | 检查源数据库是否开启 Binlog |
| 源数据库 Binlog Format 是否支持 | 检查源数据库的 binlog 格式是否为 'ROW' |
| 源数据库 binlog_row_image 是否支持 | 检查源数据库的 binlog_row_image 是否为 'FULL' |
| 目标库权限检查 | 检查 Kafka 账号是否对 Topic 有访问权限 |
| 目标库数据存在性检查 | 检查 Topic 中是否已存在数据 |