如何配置JDBC源连接器将Snowflake嵌套JSON流式传输到Kafka并输出纯JSON

一、嵌套JSON格式输出问题 #

问题背景 #

使用Snowflake的OBJECT_CONSTRUCT生成嵌套JSON，通过JDBC源连接器发送到Kafka时，最终消息是带转义的JSON字符串，而非原生JSON结构。

解决方法 #

方案1：修改Snowflake JDBC连接参数，直接识别VARIANT类型

Snowflake的OBJECT_CONSTRUCT返回VARIANT类型，JDBC驱动默认会将其序列化为字符串。在JDBC URL中添加useJsonVariant=true参数，让驱动把VARIANT类型解析为JSON对象：

jdbc:snowflake://<你的账户>.snowflakecomputing.com/?db=<数据库>&schema=<模式>&warehouse=<仓库>&role=<角色>&useJsonVariant=true

然后调整连接器配置：

"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"transforms": "unwrap",
"transforms.unwrap.type": "org.apache.kafka.connect.transforms.UnwrapFromEnvelope"

此配置下，查询返回的myRecord字段会被直接解析为JSON结构，最终Kafka消息就是目标格式：

{"id":1,"nestedField":{"nestedValue":"value"}}

方案2：使用ParseJson SMT解析字符串

如果无法修改JDBC URL，在现有ExtractField转换后添加ParseJson SMT，将提取出的JSON字符串解析为原生JSON对象：

"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"transforms": "extractField,parseJson",
"transforms.extractField.type": "org.apache.kafka.connect.transforms.ExtractField$Value",
"transforms.extractField.field": "MYRECORD",
"transforms.parseJson.type": "org.apache.kafka.connect.transforms.ParseJson$Value",
"transforms.parseJson.json.schema": ""

二、Base64解码的SMT实现 #

可以使用Kafka Connect内置的base64.Decode SMT完成解码，无需自定义代码：

• 若解码指定字段（比如字段名为encodedField）：

"transforms": "decodeBase64",
"transforms.decodeBase64.type": "org.apache.kafka.connect.transforms.base64.Decode$Value",
"transforms.decodeBase64.field": "encodedField"

• 若解码整个消息value（即value本身是base64字符串）：

"transforms": "decodeBase64",
"transforms.decodeBase64.type": "org.apache.kafka.connect.transforms.base64.Decode$Value"

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