flink cdc 连接mongo数据库。想用datastream读取整库方式。现在报错什么原因？
flink cdc 连接mongo数据库。想用datastream读取整库方式。现在报错：
Command failed with error 73 (InvalidNamespace): '{aggregate: 1} is not valid for '$changeStream'; a collection is required
MongoDBSource<String> mongoSource = MongoDBSource.<String>builder()
.hosts(Config.DB_HOST + ":" + Config.DB_PORT)
.databaseList(Config.DB_KUSER)
.collectionList()
.username(Config.DB_USER)
.password(Config.DB_PAWD)
.deserializer(Config.DEBEZIUM_DESER)
.build();
我们的Mongo版本是3.6
写回答
-
在Flink CDC中连接MongoDB数据库时，您可以使用MongoDBSource来实现。然而，根据您提供的代码和错误信息，似乎在配置collectionList时出现了问题。
根据错误信息，{aggregate: 1} is not valid for '$changeStream'; a collection is required，看起来您未指定要读取的具体集合（collection）。collectionList应该传递一个字符串集合，表示要读取的集合列表。
您需要将.collectionList()替换为.collectionList("your_collection_name")，将your_collection_name替换为您要读取的具体集合名称。如果您希望读取多个集合，可以传递一个包含所有集合名称的字符串列表。
以下是示例代码，演示如何连接MongoDB并读取指定的集合：
```MongoDBSource mongoSource = MongoDBSource.builder()
.hosts(Config.DB_HOST + ":" + Config.DB_PORT)
.databaseList(Config.DB_KUSER)
.collectionList("your_collection_name") // 指定要读取的集合名称
.username(Config.DB_USER)
.password(Config.DB_PAWD)
.deserializer(Config.DEBEZIUM_DESER)
.build();```
请确保将your_collection_name替换为您要读取的实际集合名称。这样应该能够解决报错并成功连接MongoDB数据库进行读取整库操作。
2023-10-20 14:24:31赞同 展开评论 打赏
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
实时计算Flink版是阿里云提供的全托管Serverless Flink云服务，基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎，提高作业开发运维效率。
