Flink CDC产品常见问题之flinkcdc3同步mysql到doris的时候语句不同步如何解决

问题一：Flink CDC里StarRocks是否支持oracle的外部表？怎么创建？添加到fe或be？

Flink CDC里StarRocks是否支持oracle的外部表？怎么创建？添加到fe嘛 还是be？

参考回答：

支持。

https://docs.starrocks.io/zh/docs/2.2/data_source/External_table/

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/592402

问题二：Flink CDC里为什么我在用flinkcdc3同步mysql到doris的时候语句不同步？

Flink CDC里为什么我在用flinkcdc3同步mysql到doris的时候,delete语句无法同步呢？

参考回答：

在使用 Flink CDC (Continuous Data Capture) 从 MySQL 同步数据到 Doris 时，如果 delete 语句无法同步，可能的原因有以下几点：

1. CDC Source端处理：

• 不支持Delete事件捕获：早期版本的Flink CDC可能只支持插入（INSERT）和更新（UPDATE）事件的捕获，而不支持删除事件。但Flink CDC v2.x版本集成了Debezium等工具，理论上应该能够捕获MySQL的DELETE操作。

2. 配置问题：

• 确保在配置Flink CDC连接MySQL作为数据源时，启用了binlog格式为ROW，并且设置正确以捕获所有DML操作（包括DELETE）。

3. Schema映射与转换：

• 在将变更数据流转换并写入Doris之前，检查数据转换逻辑是否正确处理了DELETE事件。例如，如果只是简单地将变更事件应用到目标表，而没有实现对Doris的相应DELETE操作，那么删除事件就不会被正确反映。

4. Doris Sink端处理：

• Doris sink connector在接收到DELETE事件后，需要能够正确地执行相应的删除操作。确认所使用的Doris sink插件或者自定义的sink逻辑是否完整实现了对DELETE事件的支持。

5. 事务和一致性保证：

• 考虑到分布式系统的一致性问题，确保在Flink作业中的事务边界设置合理，能正确处理MySQL的删除事务。

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/592407

问题三：在Flink CDC中要从检查点重启任务，如何在YAML配置文件中添加检查点的路径？

在Flink CDC中要从检查点重启任务，如何在YAML配置文件中添加检查点的路径？是否有相关文档或方法可以参考来重启Flink CDC 3.0的任务？具体怎样使用命令bin/flink run -s进行重启，应该指定哪个JAR包？

参考回答：

在 flink-conf 里设置 execution.savepoint.path 指定 savepoint 路径。

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/592409

问题四：Flink CDC里我用flink sql从kafka订阅canaljson插入数据，这是为什么？

Flink CDC里我用flink sql从kafka订阅canaljson插入数据，如果这个数据插入后，重启任务，然后发送canaljson无法删除，不重启的话就可以，为什么？

参考回答：

Flink CDC 从 Kafka 订阅 Canal JSON 格式的数据并将其插入到目标表中，如果在数据插入后重启 Flink 任务，发现无法正确处理 Canal JSON 中的 DELETE 操作，而不停止任务则可以正常处理，这个问题可能与 Flink SQL 的状态管理和 Checkpoint 机制有关。

Flink CDC 在处理 CDC 数据时，会依赖其状态来追踪和处理数据库的变更事件（包括 INSERT、UPDATE 和 DELETE）。当你重启任务时，如果没有正确设置 Checkpoint 或 Savepoint，Flink 任务重启后会从 Kafka 最新的偏移量开始消费，而忽略了之前已经消费并处理过的 DELETE 事件，因此会出现 DELETE 操作无法执行的情况。

要解决这个问题，可以采取以下措施：

1. 启用 Checkpoint：确保你的 Flink 任务启用了 Checkpoint，并且设置合理的 Checkpoint 间隔。这样在任务重启时，可以从最近的 Checkpoint 恢复状态，继续处理 Kafka 中未消费完的数据。

CREATE TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  'enable.startup.mode' = 'latest-offset',  -- 或者设置为 'group-offsets' 并确保消费组一致
  'properties.checkpoint.interval' = '60000',  -- 根据实际情况设置Checkpoint间隔
  ...
);

1. 使用 Savepoint：在任务停止前先触发一个 Savepoint，然后在重启任务时从 Savepoint 恢复，这样可以精确地恢复到任务停止前的状态。
   
2. 检查幂等性：确保下游接收系统的数据处理逻辑具有幂等性，即使同一个 DELETE 事件被处理多次，也能保证最终数据的一致性。
   

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/592411

问题五：flink cdc流计算postgresql数据库插槽可以复用吗？

flink cdc流计算postgresql数据库，数据库默认插槽数量只有32，尝试了复用插槽，将两个source表的slot.name改成相同的，会提示报错，具体报错如下图

参考回答：

如果不能复用，数据库设置几千甚至上万个插槽数量会不会带来很大的压力

关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/592788