实时计算 Flink版产品使用合集之在进行全量同步时，有两张表的数据没有正确进入，并且ID字段为null，该怎么处理

问题一：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。以下是一些基本步骤：

1. 创建 replica set：

• 在 MongoDB shell 中运行 rs.initiate() 命令以初始化一个新的 replica set。
• 将其他 mongod 实例添加到该 replica set 中，以便提高容错性和可用性。

2. 配置 Flink CDC 连接器：

• 更新 Flink CDC 连接器的配置，使其指向新的 replica set。

3. 重新启动作业：

• 重新启动你的 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呢?

参考回答：

这个问题可能是由于以下原因导致的：

1. 表结构不一致：请检查四张表的结构是否一致，包括字段名、数据类型等。如果有任何不一致，需要修改表结构以保持一致。
   
2. 数据转换问题：在使用Flink CDC进行全量同步时，可能需要对源表中的数据进行一些转换操作，例如将日期字符串转换为日期对象等。请检查这些转换操作是否正确执行。
   
3. Flink CDC配置问题：请检查Flink CDC的配置是否正确。例如，确保sink.connector设置为正确的连接器名称，如mysql-cdc。同时，检查source.table-includes和source.table-excludes配置项，以确保只同步指定的表。
   
4. 日志级别问题：请检查Flink CDC的日志级别是否设置得足够高，以便查看详细的错误信息。可以尝试将日志级别设置为DEBUG或INFO，以获取更详细的信息。
   
5. 网络问题：请检查网络连接是否正常，以及防火墙设置是否允许Flink CDC与MySQL服务器之间的通信。
   

关于本问题的更多回答可点击原文查看：

https://developer.aliyun.com/ask/574035