Kafka：从主题创建流，其值位于单独的列中

Question

我刚刚使用 postgres 源连接器将我的 kafka 连接到 postgres。现在，当我打印主题时，我得到以下输出：

rowtime: 4/1/20 4:16:12 PM UTC, key: <null>, value: {"userid": 4, "id": 5, "title": "lorem", "body": "dolor sit amet, consectetur"}
rowtime: 4/1/20 4:16:12 PM UTC, key: <null>, value: {"userid": 5, "id": 6, "title": "ipsum", "body": "cupidatat non proident"}

我如何从这个主题创建一个流，以便将值分离到它们自己的列中，就像它们最初在数据库表中一样？

额外问题：有没有办法在 jdbc-connector 中指定在创建源连接器时将列分隔到主题中？

我的连接器如下所示：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_postgres_02",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:postgresql://postgres:5432/kafka",
                "connection.user": "bob",
                "connection.password": "builder",
                "topic.prefix": "post_",
                "mode":"bulk",
                "table.whitelist" : "kafka_t.users_t",
                "poll.interval.ms" : 500
                }
        }'

Answer 1

我如何从这个主题创建一个流，以便将值分离到它们自己的列中，就像它们最初在数据库表中一样？

不是 100% 确定您的意思。例如，如果您使用 Kafka Streams，您可以创建一个带有自定义 Columns 类型（或仅使用 JSON 作为值类型）的 KStream<KeyType, Columns>，以获取数据的“列视图”。

同样，您可以将 ksqlDB 与 CREATE STREAM 命令一起使用——它可以自动将 JSON 值解析为相应的列。

额外问题：有没有办法在 jdbc-connector 中指定在创建源连接器时将列分隔到主题中？

你这是什么意思？ Kafka主题具有键值数据模型，因此如果您将任何数据存储在主题中，它必须进入键或值。如果您有更结构化的类型，例如 DB 元组，则 Kafka 代理中没有原生支持，但您需要将其放入键值模型中。