问题一：Flink中appliction 模式怎么从 checkpoint 保存的地方 启动？
Flink中appliction 模式怎么从 checkpoint 保存的地方 启动？
参考答案：
检查点是自动触发的，保存点是手动触发
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/587249
问题二：Flink CDC oracle cdc只能采集到最后一条数据，有没有遇到过的?
Flink CDC oracle cdc只能采集到最后一条数据，有没有大佬遇到过的?
参考答案：
Apache Flink CDC (Change Data Capture) 从Oracle数据库中读取数据时，如果只能采集到最后一条数据，通常是由以下几个原因导致的：
- 配置问题：
- 确保Flink CDC Oracle Connector的起始位置设置正确。默认情况下，如果未指定历史数据的处理方式和起始LSN（Log Sequence Number），可能只从当前时间点开始读取增量数据，从而只会看到最后一条变更。
- 连接器参数配置：
- 检查
debezium.oracle.log.mining.strategy等与Oracle LogMiner相关的参数是否配置正确，确保能够捕获到完整的历史变更记录。
- 表监听范围：
- 如果仅监听了某个时间点之后的数据，那么在该时间点之前的数据将不会被收集。
- 事务提交与Flink消费速度：
- 如果Flink任务消费速度非常快，以至于无法跟上Oracle数据库中的事务提交速度，且只有少量新的事务产生，那么看起来似乎只收到了最后一条数据。
- 过滤条件或分区配置：
- 检查是否存在错误的过滤条件或分区配置，这可能导致只消费了一部分数据。
- 异常中断恢复：
- 若任务曾经因故中断并重新启动，而没有使用正确的checkpoint或savepoint进行恢复，则可能只从最近的checkpoint或者重启时刻开始读取变更。
解决办法包括：
- 检查并调整Flink CDC Oracle Connector的相关配置，比如设置合适的起始SCN或时间戳。
- 确保Connector配置了正确的表名和模式，并启用全量数据扫描选项（如有需要）。
- 根据实际情况调整Flink任务的并行度和性能，使其能跟上数据库变更的速度。
- 使用恰当的checkpoint策略，并在任务重启时基于有效的checkpoint或savepoint恢复状态。
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/587230
问题三：Flink CDC如果是Mongodb，还能如何兼容jdbc关联查询效果最佳呢?
请教一下Flink CDC如果是Mongodb，除了这种trino联邦的方式，还能如何兼容jdbc关联查询效果最佳呢?
参考答案：
Apache Flink CDC 与 MongoDB 结合使用时，由于 MongoDB 不支持传统的 JDBC 接口，因此不能直接使用 JDBC 的方式进行关联查询。不过，您可以通过以下方式实现 MongoDB 数据与其它数据库（如关系型数据库）之间的近似关联查询效果：
- 使用Flink SQL Join操作：
- 如果您需要将 MongoDB 数据与其它支持 Flink CDC 的关系型数据库（如 MySQL、PostgreSQL 等）进行关联查询，可以在 Flink SQL 中使用 JOIN 语句来实现。首先确保通过 Flink CDC 从 MongoDB 和关系型数据库中读取数据流，然后在 Flink 的流处理环境中将两者进行 JOIN 操作。
-- 假设 source1 是从 MongoDB CDC 读取的数据源，source2 是从关系型数据库读取的数据源 SELECT * FROM source1 AS s1 JOIN source2 AS s2 ON s1.id = s2.foreign_key;
- 数据整合与中间存储：
- 如果需要在多个 MongoDB 集合之间或者 MongoDB 与非 CDC 支持的关系型数据库之间做关联查询，可以先将需要关联的数据抽取到一个中间存储，如 Apache Kafka 或者 Hadoop HDFS，然后在 Flink 作业中分别读取这些数据并进行 JOIN。
- 批流一体处理：
- 利用 Flink 的批流一体能力，可以将 MongoDB 的存量数据以批处理方式加载进 Flink，然后在流处理作业中实时接入 MongoDB CDC 数据，通过 Temporal Table Joins 实现实时关联查询。
- 自定义 Connector：
- 如果现有的 Flink CDC MongoDB Connector 不支持您所需要的复杂关联查询，可以考虑开发自定义的 Flink Connector，实现更灵活的数据抽取和关联逻辑。
- 数据湖方案：
- 将 MongoDB 数据和其它数据源的数据都导入到一个数据湖（如阿里云 MaxCompute、Hudi、Iceberg 等）中，然后在这个统一的数据湖上执行关联查询。
综上所述，根据您的实际业务场景和数据源特点，可以选择最适合您的方案来模拟 JDBC 关联查询的效果。在实时处理场景中，优先推荐使用 Flink SQL 的流处理能力进行 JOIN 查询。
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/587229
问题四：请教下大家Flink CDC，flinksql的这个代码要如何修改才能实现动态传参？
请教下大家Flink CDC，flinksql的这个代码要如何修改才能实现动态传参formatted-date这个参数的动态传参?实现临时表 rltest的formatted_date字段动态输出； Table rltest = tableEnv.sqlQuery(
"SELECT " +
", " +
"DATE_FORMAT(create_date, 'yyyy-MM-dd') as formatted_date " +
"FROM temp_table"
);
tableEnv.createTemporaryView("newb_temp_view", rltest);
String formattedDate ="SELECT formatted_date FROM " + rltest ; // 从 rltest 检索
tableEnv.createTemporaryView("selected_create_date",tableEnv.sqlQuery(formattedDate));
Table drrt = tableEnv.sqlQuery(
"SELECT bo.station_id , " +
"ss.station_name, " +
"COUNT(bo.id)/2 AS switch_count, " +
"ROUND(SUM(bov.switch_electric_quantity), 3)/2 AS total_switch_electric_quantity, " +
"ROUND(SUM(bo.pay_amount), 3)/2 AS total_pay_amount, " +
"'(SELECT formatted_date FROM selected_create_date)' AS create_date " +
"FROM border bo " +
"INNER JOIN \n" +
" bs_order_vehicle bov ON bo.id = bov.order_id " +
"INNER JOIN \n" +
" sys_station ss ON bo.station_id = ss.id " +
"WHERE \n" +
" bo.station_id IN (SELECT station_id FROM selected_station_id)" +
" AND DATE_FORMAT(bo.create_date, 'yyyy-MM-dd') IN (SELECT formatted_date FROM selected_create_date) " +
" AND bo.pay_type <> 0 " +
" AND bo.pay_type IS NOT NULL " +
" AND bo.order_status = 3 " +
" AND bo.order_type = 0 " +
" AND bo.del_flag = 0 " +
" GROUP BY bo.station_id, ss.station_name "
);
tableEnv.createTemporaryView("temp_table4", drrt);
Table drt = tableEnv.sqlQuery("SELECT FROM temp_table4");
tableEnv.toChangelogStream(drt).print("SQL");
env.execute();
参考答案：
要实现动态传参formatted-date，可以在查询字符串中使用占位符，然后在执行查询时传入参数。以下是修改后的代码：
String formattedDatePlaceholder = "'(SELECT formatted_date FROM selected_create_date)'"; String query = "SELECT bo.station_id, ss.station_name, COUNT(bo.id)/2 AS switch_count, ROUND(SUM(bov.switch_electric_quantity), 3)/2 AS total_switch_electric_quantity, ROUND(SUM(bo.pay_amount), 3)/2 AS total_pay_amount, " + formattedDatePlaceholder + " AS create_date FROM border bo INNER JOIN bs_order_vehicle bov ON bo.id = bov.order_id INNER JOIN sys_station ss ON bo.station_id = ss.id WHERE bo.station_id IN (SELECT station_id FROM selected_station_id) AND DATE_FORMAT(bo.create_date, 'yyyy-MM-dd') IN (SELECT formatted_date FROM selected_create_date) AND bo.pay_type <> 0 AND bo.pay_type IS NOT NULL AND bo.order_status = 3 AND bo.order_type = 0 AND bo.del_flag = 0 GROUP BY bo.station_id, ss.station_name"; Table rltest = tableEnv.sqlQuery(query); tableEnv.createTemporaryView("newb_temp_view", rltest); tableEnv.toChangelogStream(rltest).print("SQL"); env.execute();
这样，formattedDatePlaceholder变量中的占位符将被替换为实际的日期格式查询结果。
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/587228
问题五：大家有用到Flink CDC1.17.2的吗？
大家有用到Flink CDC1.17.2的吗？我提交任务到yarn上，taskmanager的metrics没有数据 
参考答案：
Flink CDC 1.17.2 是一个用于捕获数据库变更的框架，它可以将数据库中的变更数据实时同步到 Flink 中。关于您提到的问题，任务提交到 YARN 上后，taskmanager 的 metrics 没有数据，可能有以下几种原因：
- 配置问题 ：请检查您的 Flink CDC 配置文件（如
flink-conf.yaml），确保相关配置项已正确设置。例如，检查以下配置项：
# 设置 Flink CDC 的 Checkpoint 间隔时间 execution.checkpointing.interval: 5000 # 设置 Flink CDC 的 Checkpoint 超时时间 execution.checkpointing.timeout: 300000 # 设置 Flink CDC 的 TaskManager 数量 taskmanager.numberOfTaskSlots: 4
- 资源不足 ：如果 taskmanager 的资源不足，可能导致任务无法正常运行。请检查您的 YARN 集群资源分配情况，确保有足够的资源供任务使用。
- 日志问题 ：查看 Flink CDC 和 TaskManager 的日志文件，以获取更多关于问题的详细信息。这有助于诊断问题并找到解决方案。
- 版本兼容性问题 ：请确保您使用的 Flink CDC 版本与您的 Flink 和 YARN 集群兼容。您可以查阅 Flink CDC 官方文档，了解支持的版本信息。
关于本问题的更多回答可点击进行查看：
