Kafka Connect-JSON转换器-JDBC接收器连接器-列类型JSON

Question

用例是将整个消息（JSON）和键作为记录存储在表中，表中有两列“id”和“data”。

数据库是Postgres，它支持JSON的列类型。

根据本文，JSONCon的支持类型是字符串、int64等https://cwiki.apache.org/confluence/display/KAFKA/KIP-301：JsonConver的模式推理

是否可以将数据字段的类型设置为JSON，然后将其存储在Postgres DB中，列类型为JSON。

schema = `{
"type":"struct",
"fields":[
    {"type":"string", "optional": false, "field":"id"},
    {"type":"string", "optional": false, "field":"data"}
]}`

样本数据有效载荷是

"payload": { "id": 10000, "data": {"hello":"world"} }

上面将数据存储为文本，并期望列在Postgres中为文本类型。如果Postgres上的列是JSON类型，那么JDBCSink连接器将抛出错误。

在Postgres上使用JSON类型将有助于在JSON字段等上创建索引。是否可以适当地使用JSON转换器和JDBC Sink转换器来存储列类型为JSON的记录。

Answer 1

JDBC Sink连接器不支持PostgreSQL json、jsonb类型。它支持原始类型、日期时间的数量。

在以下页面中，您可以找到将模式类型映射到数据库类型(PostgreSQL)https://docs . confluent . io/5 . 1 . 0/connect/Kafka-connect-JDBC/sink-connector/index . html

虽然 JDBC 源连接器在某些部分支持 json、jsonb 类型 - 这种类型的列不会映射到 STRUCT，但会映射到 STRING 类型。

Answer 2

使用value.converter.schema。enable=true，并像这样发送JSON数据（将模式作为每条消息的一部分，并用实际消息数据更新payload部分），它应该与JDBCSink一起工作。

{
    "schema": {
        "type": "struct",
        "fields": [{
            "type": "int32",
            "optional": false,
            "field": "id"
        }, {
            "type": "struct",
            "name": "data",
            "optional": false,
            "fields": [{
               "type": "string",
               "name": "hello",
               "optional":false
            }]
        }],
        "optional": false,
        "name": "foobar"
    },
    "payload": {
        "id": 10000,
        "data": {"hello":"world"}
    }
}

请注意，字段

或者你可以考虑转换你的客户使用Avro，为自己节省一些网络带宽。