问题一：Flink CDC里 orace数据库增删改，想同步到pg数据库，有什么好的方式吗？
Flink CDC里 orace数据库增删改，想同步到pg数据库，有什么好的方式吗？
参考答案：
Flink Change Data Capture (CDC) 提供了一种方法可以从Oracle数据库捕获数据更改，并将其实时同步到PostgreSQL数据库。以下是使用Flink CDC实现Oracle到PostgreSQL数据同步的基本步骤和注意事项：
1. 配置Oracle端：
在Oracle数据库上启用CDC功能，这通常涉及设置额外的日志模式和启用数据库触发器或使用GoldenGate等工具来捕获更改。
确保Oracle数据库产生的日志足以包含所有需要同步的更改。
2. 安装和配置Flink CDC：
使用Flink CDC的Oracle connector，它可以消费Oracle的redo logs或通过GoldenGate等工具提供的接口获取变更数据。
下载并添加适用于Oracle的Flink CDC连接器到你的Flink环境中。
3. 创建Flink CDC作业：
创建一个新的Flink流处理作业，定义Oracle作为source，使用对应的Oracle CDC connector指定数据库连接信息、表列表等。
设置数据转换逻辑（如果有类型映射、字段裁剪或转换需求）。
4. 配置PostgreSQL sink：
使用Flink的PostgreSQL sink connector将变更数据写入到PostgreSQL数据库中。
确保PostgreSQL数据库已准备好接收数据，包括表结构匹配、权限设置等。
考虑事务一致性问题，特别是在并发插入、更新和删除操作时，可能需要在sink端采取特定策略来保持一致性。
5. 性能优化：
根据数据量和业务需求调整Flink作业的并行度。
配置合适的buffer size和batch size以优化网络传输和写入性能。
如果遇到数据格式或编码问题（如中文乱码），确保两端数据库字符集一致或适配。
6. 监控和调试：
监控作业状态和性能指标，及时发现并解决可能出现的问题，比如网络延迟、数据积压等。
7. 故障恢复与幂等性：
设计作业以支持故障恢复，保证即使在Flink作业重启后也能从上次成功的位置继续同步。
如果有必要，考虑在sink端实现幂等性写入，防止重复数据写入。
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/609937
问题二：flink cdc3.0.1连接到oracle 但是读取不到数据怎么办？
flink cdc3.0.1连接到oracle 但是读取不到数据怎么办？
参考答案：
当使用 Apache Flink CDC 3.0.1 连接到 Oracle 数据库但无法读取到数据时，可以按照以下步骤排查问题：
1. 配置验证：
确认Flink CDC作业中对Oracle数据库的连接配置是否正确，包括但不限于：
数据库URL、端口、服务名或SID
用户名和密码
驱动类路径
是否正确设置了Oracle CDC所需的表或SCN范围
确保使用的Oracle JDBC驱动与Flink CDC兼容，并且已放置在Flink的lib目录下。
2. 数据库权限：
检查用于Flink CDC的Oracle账户是否具有足够的权限去读取所需的表和查看相关变更日志。一般而言，需要至少具有查询和READ CHANGE DATA权限。
3. 数据库CDC设置：
确认Oracle数据库启用了适当的日志级别，以便能够捕获变更数据。
如果使用的是LogMiner或GoldenGate等Oracle CDC机制，请确认其配置正确有效。
4. 网络和连接测试：
检查Flink CDC作业运行节点与Oracle数据库之间的网络连通性，确保没有任何防火墙或安全组规则阻止连接。
尝试直接从运行Flink任务的节点通过命令行或应用程序连接Oracle数据库，验证基本的数据库连接。
5. 任务状态和日志分析：
查看Flink CDC任务的运行日志，查找任何有关连接失败或读取数据错误的信息。
分析Flink UI或日志中是否有具体的错误提示，例如是否存在表找不到、列映射问题、数据转换异常等情况。
6. 时间区域问题：
若之前存在时间区域相关问题的历史记录，确认系统和JDBC驱动的时间区域设置与Oracle数据库一致。
7. Flink CDC版本与Oracle兼容性：
确保使用的Flink CDC版本与所连接的Oracle数据库版本兼容，有时新版本的Flink CDC可能尚未支持老版本的Oracle数据库。
8. 数据活动检查：
- 确认Oracle数据库中有实际的数据变更发生。如果没有新增、修改或删除操作，自然不会有任何CDC数据可供读取。
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/609930
问题三：Flink CDC里为什么会这样？
flink 机器所在的时区或者mysql机器所在时区不对，所以有8小时问题。如果是 sink 写出来的时区和期望不一致，可以在pipeline 下面添加 local-time-zone 参数去修改写出数据对应的时区。
参考答案：
dateTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();解析代码写死的UTC时间o.debezium.time.Timestamp ，这个类，改好后替换原jar
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/609925
问题四：Flink CDC里请问这句话怎么理解？
Flink CDC里请问云邪老师在FFA上这里讲的“框架在从全量阶段切换到增量阶段时，会自动地把并发度调低”，会把空闲的taskmanager回收么？是cdc框架给重启作业调整并行度么？https://flink-learning.org.cn/article/detail/3df74aa43599938de19ab1e903001cfe 
参考答案：
scan.incremental.close-idle-reader.enabled这个参数控制的。 
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/609921
问题五：请教下 flink-cdc同步到kafka怎么才能拿到完整热搜canal-json格式？
请教下 flink-cdc同步到kafka怎么才能拿到完整热搜canal-json格式，现同步到kafka都是这种格式：{"data":[{"id":1,"cl1":363,"cl2":1}],"type":"INSERT"} 没有mysqltype database那些scheam信息flink-cdc输出到kafka热搜debezium-json格式好像也是没有schema信息。{"before":null,"after":{"id":1,"cl1":363,"cl2":1},"op":"c"}都是这种。难道要用api去解析入kafka？我现在用的flink sql 在想是不是有什么参数可以打开写入scheam？
参考答案：
flink-cdc 拿到的提供的数据格式是debezium-json。canal-json格式的需要自己去序列化格式。参考：
https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-connector-properties
。
关于本问题的更多回答可点击进行查看：
