【发布时间】:2021-09-06 01:40:51
【问题描述】:
我一直在为 PostgreSQL 实现一个 kafka 连接器(我正在使用 debezium kafka 连接器并通过 docker 运行所有部分)。我需要一个自定义分区键,所以我一直在使用 SMT 来实现这一点。但是,我使用的方法创建了一个结构,我需要它是一个字符串。这个article 贯穿了如何将分区键设置为int,但我无法访问配置文件来设置适当的转换。目前我的kafka连接器是这样的
curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
"name": "connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"tasks.max": "1",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "postgres",
"database.password": "password",
"database.dbname" : "postgres",
"database.server.name": "postgres",
"table.include.list": "public.table",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "schema-changes.table",
"transforms": "routeRecords,unwrap,createkey",
"transforms.routeRecords.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.routeRecords.regex": "(.*)",
"transforms.routeRecords.replacement": "table",
"transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
"transforms.createkey.type": "org.apache.kafka.connect.transforms.ValueToKey",
"transforms.createkey.fields": "id"
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter": "org.apache.kafka.connect.storage.StringConverter"
}
}'
我知道我必须提取列的值,但我不确定如何。
【问题讨论】:
-
请告诉我们您尝试提取列的值
-
@Todd 是上述配置 sn-p
"transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState", "transforms.createkey.type": "org.apache.kafka.connect.transforms.ValueToKey", "transforms.createkey.fields": "id"中的unwrap和createKey转换。但就像我上面所说的,这会创建一个具有id值的结构。但我需要密钥只能是字符串值
标签: apache-kafka apache-kafka-connect