问题一：Flink CDC实时采集mongodb，是不是只支持mongo集群？
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 集群上使用。这是因为 MongoDB 的 Change Streams 功能（用于捕获数据变更）仅在 replica set 或者 sharded cluster 中可用。
如果你正在尝试从单节点 MongoDB 实例中收集数据，你需要将其转换为一个 replica set 或者 sharded cluster。以下是一些基本步骤：
- 创建 replica set：
- 在 MongoDB shell 中运行
rs.initiate()命令以初始化一个新的 replica set。 - 将其他 mongod 实例添加到该 replica set 中，以便提高容错性和可用性。
- 配置 Flink CDC 连接器：
- 更新 Flink CDC 连接器的配置，使其指向新的 replica set。
- 重新启动作业：
- 重新启动你的 Flink CDC 作业，它现在应该能够成功地从 MongoDB replica set 中读取数据了。
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/574040
问题二：Flink CDC连kafka的时候，如果消费掉的消息，能删除掉吗?
Flink CDC连kafka的时候，如果消费掉的消息，能删除掉吗?
参考回答：
是的，Flink CDC在连接到Kafka时，可以消费掉的消息并将其删除。这是因为Flink CDC会监控并捕获数据库的变动，包括数据的插入、更新和删除，然后将这些变更按发生的顺序完整记录下来，写入到消息中间件中以供其他服务进行订阅及消费。当Flink CDC消费掉Kafka中的消息后，它会自动删除这些消息，以防止重复消费。
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/574038
问题三：Flink CDC 可以读取虚拟列op吗？
Flink CDC 可以读取虚拟列op吗？想基于op，进行不同的数据操作
参考回答：
Flink CDC 可以读取虚拟列 op，但是需要使用自定义的 DebeziumDeserializationSchema 来解析该列。
具体来说，您需要在 DebeziumDeserializationSchemaBuilder 中添加一个名为 op 的列，并指定其数据类型和反序列化器。例如：
DebeziumDeserializationSchema<MyEvent> deserializationSchema = DebeziumDeserializationSchemaBuilder<MyEvent>.create() .schemaName("my_schema") .databaseList(Arrays.asList("mydb")) .tableList(Arrays.asList("mytable")) .column("id", DataTypes.BIGINT()) .column("name", DataTypes.STRING()) .column("op", DataTypes.STRING()) // 添加虚拟列 op .deserializer(new MyEventDeserializer()) .build();
然后，您可以在您的应用程序中使用 FlinkCDC 连接器来读取数据流，并根据 op 列的值执行不同的操作。例如：
FlinkCDC<MyEvent> cdc = new FlinkCDC<>(env, mySourceFunction, deserializationSchema); DataStream<MyEvent> stream = env.addSource(cdc); stream.filter(event -> event.getOp().equals("INSERT")) // 根据 op 列的值过滤数据流 .map(event -> { /* 对 INSERT 事件执行某些操作 */ return event; }) .otherwise(event -> { /* 对其他事件执行某些操作 */ return event; });
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/574037
问题四：我使用flinkcdc的sql形式进行全量同步，4张表，有两张表数据没进去，，id怎么是null呢?
我使用flinkcdc的sql形式进行全量同步，4张表，有两张表数据没进去，看日志，id怎么是null呢? 
参考回答：
这个问题可能是由于以下原因导致的：
- 表结构不一致：请检查四张表的结构是否一致，包括字段名、数据类型等。如果有任何不一致，需要修改表结构以保持一致。
- 数据转换问题：在使用Flink CDC进行全量同步时，可能需要对源表中的数据进行一些转换操作，例如将日期字符串转换为日期对象等。请检查这些转换操作是否正确执行。
- Flink CDC配置问题：请检查Flink CDC的配置是否正确。例如，确保
sink.connector设置为正确的连接器名称，如mysql-cdc。同时，检查source.table-includes和source.table-excludes配置项，以确保只同步指定的表。 - 日志级别问题：请检查Flink CDC的日志级别是否设置得足够高，以便查看详细的错误信息。可以尝试将日志级别设置为
DEBUG或INFO，以获取更详细的信息。 - 网络问题：请检查网络连接是否正常，以及防火墙设置是否允许Flink CDC与MySQL服务器之间的通信。
关于本问题的更多回答可点击原文查看：
