MySQL JSON列在Kafka Connector中无法扁平化的问题求助

方法一：通过Kafka Connector的Transforms直接处理JSON字段 #

如果使用Debezium MySQL源连接器（主流的MySQL-Kafka同步组件），可以通过内置转换功能提取JSON字段中的键，将其扁平化到顶层字段。

1. 配置源连接器的Transforms #

在Debezium源连接器的配置中添加以下参数（Properties格式示例）：

# 基础配置（省略数据库地址、账号等通用项）
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=your-source-db-host
database.port=3306
database.user=your-user
database.password=your-pass
database.server.id=101
database.server.name=source-db
table.include.list=your_db.your_table

# 启用转换规则
transforms=unwrap,flattenMeta
# 提取新记录状态（移除Debezium的事件包装层）
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
# 使用自定义脚本处理JSON扁平化
transforms.flattenMeta.type=org.apache.kafka.connect.transforms.Script
# 指定Groovy脚本路径（确保Connector可访问该文件）
transforms.flattenMeta.script=flatten_meta.groovy

2. 编写Groovy转换脚本（flatten_meta.groovy） #

脚本负责提取meta JSON字段中的first、middle、last键，生成扁平化的顶层字段：

def meta = value.get("meta")
if (meta != null) {
    // 将JSON键映射为新的顶层字段
    value.put("meta_first", meta.get("first"))
    value.put("meta_middle", meta.get("middle"))
    value.put("meta_last", meta.get("last"))
    // 可选：若无需保留原meta字段，可执行移除
    // value.remove("meta")
}
return value

3. 目标端连接器配置 #

目标MySQL连接器只需将Kafka消息中的扁平化字段（如meta_first）映射到目标表对应列即可，字段名一致时可直接同步。

方法二：在源MySQL端创建视图提前扁平化 #

如果不想在Connector层处理，可在源数据库创建视图，直接将JSON字段展开为普通列，再让Kafka Connector同步该视图。

1. 创建扁平化视图 #

在源MySQL中执行SQL创建视图：

CREATE VIEW flattened_your_table AS
SELECT 
    id, -- 假设表有主键id
    JSON_UNQUOTE(JSON_EXTRACT(meta, '$.first')) AS meta_first,
    JSON_UNQUOTE(JSON_EXTRACT(meta, '$.middle')) AS meta_middle,
    JSON_UNQUOTE(JSON_EXTRACT(meta, '$.last')) AS meta_last
FROM your_table;

注意：MySQL 8.0+版本可使用更简洁的meta->>'$.first'替代JSON_UNQUOTE(JSON_EXTRACT(...))

2. 配置Kafka Connector同步视图 #

修改Debezium源连接器的table.include.list参数，将视图加入同步范围：

table.include.list=your_db.flattened_your_table

后续目标端连接器直接同步该视图的普通列即可，无需处理JSON类型。

内容的提问来源于stack exchange，提问作者Ravi Ranjan Pandey