如何通过MongoDB Sink Connector按Kafka主题分组到不同库和集合

你提到的FieldPathNamespaceMapper正是实现需求的正确工具，它支持通过Kafka主题名的结构动态拆分出MongoDB的数据库名和集合名，完美解决跨数据库隔离数据的问题。以下是具体配置步骤和示例：

核心原理 #

Debezium Postgres源连接器默认生成的主题格式为{前缀}.{数据库实例名}.{schema名}.{表名}（如pg-source.inventory.products），我们可以通过正则表达式捕获主题中的数据库和表名部分，直接映射为MongoDB的数据库.集合命名空间。

关键配置参数 #

针对你的需求，需要配置以下核心参数：

• namespace.mapper：指定使用FieldPathNamespaceMapper处理命名空间映射
• field.path.namespace.mapper.topic.regex：正则表达式，捕获主题名中的数据库和集合标识
• field.path.namespace.mapper.topic.replacement：将捕获的内容拼接为数据库名.集合名格式

完整Sink连接器配置示例 #

假设你的Debezium主题格式为pg-source.{db-name}.{table-name}（如pg-source.inventory.products、pg-source.order_db.orders），配置如下：

name=mongo-multi-db-sink
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=2
# 匹配所有需要同步的Debezium主题，或用topics.regex批量匹配
topics=pg-source.inventory.products,pg-source.order_db.orders
connection.uri=mongodb://mongodb-user:password@localhost:27017/

# 启用FieldPathNamespaceMapper
namespace.mapper=com.mongodb.kafka.connect.sink.namespace.mapping.FieldPathNamespaceMapper

# 正则捕获主题中的数据库名和表名
# 正则解释：匹配"pg-source."后的第一个单词（数据库名）和第二个单词（表名）
field.path.namespace.mapper.topic.regex=^pg-source\\.(\\w+)\\.(\\w+)$
# 将捕获组替换为MongoDB的命名空间格式：数据库.集合
field.path.namespace.mapper.topic.replacement=$1.$2

# 其他必要配置
document.id.strategy=com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy
document.id.strategy.partial.value.path=id  # 用源数据的id作为MongoDB文档id
delete.on.null.values=true  # 处理Debezium的删除事件

配置调整说明 #

如果你的Debezium主题命名格式不同，只需修改正则表达式即可：

• 若主题格式为{db-name}.{table-name}（如inventory.products），正则改为：

  field.path.namespace.mapper.topic.regex=^(\\w+)\\.(\\w+)$
  field.path.namespace.mapper.topic.replacement=$1.$2
  

• 若主题包含更多层级（如pg-cluster-1.inventory.public.products），正则调整为捕获对应位置的分组即可。

注意事项 #

1. 确保MongoDB连接用户拥有创建数据库和集合的权限（如readWriteAnyDatabase或dbAdmin角色），连接器会自动创建不存在的数据库和集合。
2. 如果需要同步大量主题，可使用topics.regex替代topics，比如topics.regex=pg-source-.*\\..*\\..*匹配所有符合格式的主题。
3. 对比topic.regex：topic.regex仅能将匹配主题映射到单个数据库下的不同集合，而FieldPathNamespaceMapper支持跨数据库的动态映射，完全满足你的隔离需求。

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