Flink CDC产品常见问题之flink postgresqlcdc 报错如何解决

问题一：Flink CDC里我参照官网CDC3.0案例，MySQL整库同步到Doris怎么做？

Flink CDC里我参照官网CDC3.0案例，MySQL整库同步到Doris

bash bin/flink-cdc.sh mysql-to-doris.yaml

然后看到群里有大佬提到新增表可以先创建Savepoint，然后再从savepoint重启。

想请问一下，flink-cdc.sh 这种作业，要从savepoint重启的命令是怎么样的？

Flink从savepoint的命令在此处不适用吧？conf/flink-conf.yaml 中有添加参数：state.savepoints.dir: hdfs://xxx/flink/flink-savepoints

然后我也创建了savepoint 文件并停止了作业：./bin/flink stop --savepointPath my_job_id

但是现在要如何把作业重新启动，就不知道咋整了？

参考答案：

在flink-conf里面手动添加下路径，3.1会把这个参数暴露出来。pipeline，目前不能动态加表，这个参数灭有暴露出来。

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

https://developer.aliyun.com/ask/599297?spm=a2c6h.12873639.article-detail.12.50e24378TRW91E

问题二：flink CDC处理数据，记录每条数据的变更 大家是用什么技术实现的？

flink CDC处理数据，记录每条数据的变更 大家是用什么技术实现的？

参考答案：

在Flink CDC中处理数据并记录每条数据的变更，通常是通过Change Data Capture（CDC）技术实现的。

Flink CDC是一个基于数据库日志的实时数据集成框架，它能够捕获源数据库的增量变动记录，并将这些变动同步到一个或多个数据目的地。具体来说，实现这一功能的关键技术和步骤包括：

1. 使用CDC工具：Flink CDC利用像Debezium这样的CDC工具来采集源数据库的变更日志。这些工具能够识别数据库中的插入、更新、删除等操作，并将这些操作的数据转换成Flink可以处理的格式。
2. RowData与RowKind：Flink SQL内部支持了完整的changelog机制。通过重构的TableSource接口，Flink CDC可以将CDC数据转换成Flink认识的格式。在Flink中，每行数据被表示为RowData数据结构，并且附带有RowKind元数据信息，这些信息包括了插入、更新前、更新后、删除等操作类型，这与数据库中的binlog概念类似。
3. schema变更自动同步：Flink CDC 3.0提供了schema变更自动同步的功能，这意味着当数据源发生schema变更时，用户无需手动介入，系统会自动同步这些变更，从而大大降低了用户的运维成本。
4. 全增量一体化与高级特性：Flink CDC支持全增量一体化、无锁读取、并行读取等高级特性，配合Flink优秀的管道能力和丰富的上下游生态，可以实现高效的数据处理和同步。
5. 分布式架构：Flink CDC采用分布式架构，可以在大规模数据集和复杂的数据集成场景中发挥作用，提供高性能和高可靠性的数据同步能力。

综上所述，Flink CDC通过结合CDC技术和Flink自身的数据处理能力，实现了对数据库变更的高效捕获和同步。这些技术的应用使得Flink CDC成为了一个强大的实时数据集成工具，适用于多种数据同步和集成场景。

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

https://developer.aliyun.com/ask/599296?spm=a2c6h.12873639.article-detail.13.50e24378TRW91E

问题三：Flink CDC里大家有用 oracle-cdc使用xstream 采集 oracle 的吗？

Flink CDC里大家有用 oracle-cdc使用xstream 采集 oracle 的吗？我这边使用这种方式采集 oracle，存量数据可以采集过来，但是后面数据更新的数据无法捕捉到。开了ck，是怎么从checkpoint 开始重跑的？

参考答案：

在Flink CDC中使用oracle-cdc和xstream采集Oracle数据时，如果遇到存量数据可以采集过来，但后续的数据更新无法捕捉到的问题，这通常与几个关键配置有关。以下是一些可能的原因和解决方案：

1. 确保Oracle数据库配置正确：需要确认Oracle数据库已经开启了归档日志和补充日志，因为Flink CDC基于Debezium的LogMiner实现，而LogMiner需要使用这些日志来获取变更数据。
2. 检查Flink CDC的配置：确保Flink CDC的配置正确，特别是与checkpoint相关的配置。正确的配置可以帮助Flink CDC从checkpoint开始重跑，捕捉到数据更新。
3. 使用XStream模式的注意事项：如果您使用的是XStream模式，需要注意的是，XStream模式只能获取到自启动以来的增量数据，无法获取到存量数据。因此，如果需要存量数据，可能需要使用其他工具如Oracle GoldenGate来进行同步。
4. 版本兼容性：确保您使用的Flink CDC版本与Oracle数据库版本兼容。Flink CDC 2.1版本通过引入内置Debezium组件，增加了对Oracle的支持。
5. 监控和日志：检查Flink CDC的监控和日志信息，这些信息可以帮助您定位问题所在。例如，查看是否有错误日志提示，或者Flink CDC是否有反压的情况发生。
6. 重启作业：如果上述配置都正确无误，您可以尝试停止当前的Flink CDC作业，然后重新创建savepoint并从savepoint重启作业，看是否能够捕捉到数据更新。

综上所述，如果问题依旧存在，建议查阅Flink CDC的官方文档或寻求社区支持，以获得更具体的帮助。同时，也可以考虑在Flink社区论坛或相关技术交流群中提问，那里有许多经验丰富的开发者和用户，他们可能会提供有用的建议或解决方案。

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

https://developer.aliyun.com/ask/599295?spm=a2c6h.12873639.article-detail.14.50e24378TRW91E

问题四：现在Flink CDC-3.0.1支持从PG(16.x)的备库复制嘛？

现在Flink CDC-3.0.1支持从PG(16.x)的备库复制嘛？

参考答案：

目前不支持。3.1快发了可以试试，我这边还没做测试，debezium那边说支持，

https://github.com/ververica/flink-cdc-connectors/pull/3034

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

https://developer.aliyun.com/ask/599294?spm=a2c6h.12873639.article-detail.15.50e24378TRW91E

问题五：Flink CDC里flink postgresqlcdc 报错，这个是什么原因导致的？

Flink CDC里flink postgresqlcdc 报错，这个是什么原因导致的？ 找了资料说是 pg 9.5 以上才支持 Debezium的，我现在pg库就已经是15的了

参考答案：

试试3.0。

https://stackoverflow.com/questions/59634983/kafka-connect-error-unable-to-obtain-valid-replication-slot

, 看debezium代码是查了900次都没查到(select * from pg_replication_slots where slot_name = ? and database = ? and plugin = ?)，

https://github.com/debezium/debezium/blob/71c2084c4b596fb8fcb118225196730cbb4fa509/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java#L334

，如果3.0还不行建议咨询一下数据库团队是不是卡住了。

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

https://developer.aliyun.com/ask/599293?spm=a2c6h.12873639.article-detail.16.50e24378TRW91E