2023-11-24 06:56:17,588 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: ftp_logs[1] (1/1) (00a811d7e67b1745da551cb75b99613e_bc764cd8ddf7a0cff126f51c16239658_0_37) switched from RUNNING to FAILED on 10.44.0.16:40629-417690 @ 10.44.0.16 (dataPort=37924).
com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.errors.ConnectException: com.mongodb.MongoCommandException: Command failed with error 40573 (Location40573): 'The $changeStream stage is only supported on replica sets' on server 10.191.20.100:27017. The full response is {"ok": 0.0, "errmsg": "The $changeStream stage is only supported on replica sets", "code": 40573, "codeName": "Location40573"}
at com.mongodb.kafka.connect.source.MongoSourceTask.setCachedResultAndResumeToken(MongoSourceTask.java:566) ~[flink-sql-connector-mongodb-cdc-2.3.0.jar:2.3.0]
at com.mongodb.kafka.connect.source.MongoSourceTask.start(MongoSourceTask.java:196) ~[flink-sql-connector-mongodb-cdc-2.3.0.jar:2.3.0]
at com.ververica.cdc.connectors.mongodb.internal.MongoDBConnectorSourceTask.start(MongoDBConnectorSourceTask.java:101) ~[flink-sql-connector-mongodb-cdc-2.3.0.jar:2.3.0]
at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:759) ~[flink-sql-connector-mongodb-cdc-2.3.0.jar:2.3.0]
Flink CDC实时采集mongodb,是不是只支持mongo集群?
Flink CDC实时采集MongoDB时,确实只支持MongoDB的副本集(replica sets)。这是因为$changeStream
阶段是MongoDB的副本集特性,用于捕获数据变更。如果使用非副本集的MongoDB实例,将无法使用Flink CDC进行实时数据采集。
MongoDB CDC 连接器支持通过副本集或分片集架构模式读取阿里云云数据库MongoDB版的数据,也支持读取自建MongoDB数据库的数据 。MongoDB CDC连接器的支持信息如下:https://help.aliyun.com/zh/flink/developer-reference/mongodb-cdc-connector?spm=a2c4g.11186623.0.i220
是的,Flink CDC 实时采集 MongoDB 数据时,只支持在 MongoDB 集群上使用。这是因为 MongoDB 的 Change Streams 功能(用于捕获数据变更)仅在 replica set 或者 sharded cluster 中可用。
如果你正在尝试从单节点 MongoDB 实例中收集数据,你需要将其转换为一个 replica set 或者 sharded cluster。以下是一些基本步骤:
创建 replica set:
rs.initiate()
命令以初始化一个新的 replica set。配置 Flink CDC 连接器:
重新启动作业:
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。
实时计算Flink版是阿里云提供的全托管Serverless Flink云服务,基于 Apache Flink 构建的企业级、高性能实时大数据处理系统。提供全托管版 Flink 集群和引擎,提高作业开发运维效率。