Kafka JDBC Sink：能否将Kafka Topic值存储为表中单个列的值？

要把Kafka消息中整个value（JSON格式）以字符串形式存入数据库的单列，你当前的配置问题在于JsonConverter会自动解析JSON结构，而非保留原始字符串。可以通过以下两种方案实现需求：

方案一：直接使用StringConverter传递原始JSON字符串 #

这是最直接的方式，让转换器把整个value当作纯字符串处理，不做解析。

修改后的转换器配置 #

"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"key.converter.schemas.enable":"false",
"value.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter.schemas.enable":"false"

补充JDBC Sink核心配置 #

假设目标数据库表有一个名为json_payload的字符串类型列（如MySQL的TEXT、PostgreSQL的TEXT），添加以下配置完成字段映射：

// 数据库连接配置
"connection.url": "jdbc:mysql://your-db-host:3306/your-db-name",
"connection.user": "db-user",
"connection.password": "db-password",

// 目标表与写入模式
"table.name.format": "your_target_table",
"insert.mode": "insert", // 若需更新则用"upsert"，配合主键配置

// 字段映射：将Kafka的value映射到目标列json_payload
"fields": "value:json_payload"

如果需要基于key做更新操作，需额外添加主键配置：

"pk.mode": "record_key",
"pk.fields": "id" // 对应数据库表的主键字段

方案二：使用JsonConverter配合SMT包装为单字段 #

若因其他需求必须保留JsonConverter，可通过单消息转换（SMT）把整个JSON结构包装成一个字段，再映射到数据库列。

完整配置示例 #

"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"key.converter.schemas.enable":"false",
"value.converter":"org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable":"false",

// SMT配置：将整个value包装为名为json_payload的字段
"transforms": "wrapValue",
"transforms.wrapValue.type": "org.apache.kafka.connect.transforms.WrapIntoSingleMessageField$Value",
"transforms.wrapValue.field": "json_payload",

// JDBC Sink核心配置
"connection.url": "jdbc:mysql://your-db-host:3306/your-db-name",
"connection.user": "db-user",
"connection.password": "db-password",
"table.name.format": "your_target_table",
"insert.mode": "insert"

注意事项 #

• 确保目标数据库列的类型支持存储长字符串，避免因长度限制导致写入失败。
• 若使用upsert模式，需保证数据库表已创建对应主键，且pk.fields配置正确。

内容的提问来源于stack exchange，提问作者SVMevha