Mongo Source Kafka Connector如何将数组拆分为单条记录输出到Kafka Topic

实现思路 #

• 你可以直接使用Kafka Connect内置的**单消息转换（SMT）**能力即可完成数组拆分和字段打平，不需要修改连接器核心逻辑，仅需在现有配置基础上追加SMT相关配置即可实现需求。

配置修改说明 #

你需要在原有Connector配置的config块中追加以下配置项，完成两层数组拆分和字段打平：

"transforms": "explodeContacts,explodeNumbers,flatten",
"transforms.explodeContacts.type": "org.apache.kafka.connect.transforms.Explode$Value",
"transforms.explodeContacts.field": "contacts",
"transforms.explodeNumbers.type": "org.apache.kafka.connect.transforms.Explode$Value",
"transforms.explodeNumbers.field": "contacts.contact_numbers",
"transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
"transforms.flatten.delimiter": "_"

同步修改output.schema.value配置，适配打平后的字段名：

"output.schema.value":"{\"name\":\"MongoExchangeSchema\",\"type\":\"record\",\"namespace\":\"com.mongoexchange.avro\",\"fields\":[{\"name\": \"client_id\",\"type\": \"string\"},{\"name\": \"device_id\",\"type\": \"string\"}, {\"name\": \"contacts_contact_name\",\"type\": \"string\"}, {\"name\": \"contacts_contact_numbers_contact_num\",\"type\": \"string\"}]}"

完整的创建Connector命令 #

curl -X POST -H "Content-Type: application/json" --data '
  {"name": "mongo-source-contacts",
   "config": {
     "tasks.max":"1",
     "connector.class":"com.mongodb.kafka.connect.MongoSourceConnector",
     "output.format.value":"schema",
     "output.schema.value":"{\"name\":\"MongoExchangeSchema\",\"type\":\"record\",\"namespace\":\"com.mongoexchange.avro\",\"fields\":[{\"name\": \"client_id\",\"type\": \"string\"},{\"name\": \"device_id\",\"type\": \"string\"}, {\"name\": \"contacts_contact_name\",\"type\": \"string\"}, {\"name\": \"contacts_contact_numbers_contact_num\",\"type\": \"string\"}]}",
     "schema.compatibility": "NONE",
     "key.converter":"org.apache.kafka.connect.storage.StringConverter",
     "value.converter":"io.confluent.connect.avro.AvroConverter",
     "value.converter.schema.registry.url":"http://localhost:8081",
     "connection.uri":"mongodb://localhost:27017/replicaSet=globaldb",
     "publish.full.document.only": true,
     "topic.prefix":"t_cb",
     "topic.creation.default.partitions": 4,
     "topic.creation.default.replication.factor": 1,
     "database":"testdb",
     "collection":"contactbook_test",
     "transforms": "explodeContacts,explodeNumbers,flatten",
     "transforms.explodeContacts.type": "org.apache.kafka.connect.transforms.Explode$Value",
     "transforms.explodeContacts.field": "contacts",
     "transforms.explodeNumbers.type": "org.apache.kafka.connect.transforms.Explode$Value",
     "transforms.explodeNumbers.field": "contacts.contact_numbers",
     "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
     "transforms.flatten.delimiter": "_"
}}' http://localhost:8083/connectors -w "\n"

输出效果 #

你给出的示例文档最终会输出2条记录到Kafka Topic，完全符合你要求的单行数据格式：

1. 字段值对应：11 11aa11 FirstName LastName +4912222222
2. 字段值对应：11 11aa11 FirstName2 LastName2 +4911111111

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