问题一：flink CDC 增量同步的问题 第一个job 同步有较高的延迟，延迟时间 5~40 分钟不等？
flink CDC 增量同步的问题
一个job运行之后，刚开始同步很快，几乎秒级，再提交一个job 之后，第一个job 同步有较高的延迟，延迟时间 5~40 分钟不等？ 
参考回答：
Flink CDC 在处理增量同步时，其默认策略是首次运行时进行全量同步，随后便转为增量读取。这种机制对于表数据量较大的情形特别重要，因此在重写 AbstractMessageListener#read 方法以适应大量数据处理的场景时需要格外小心。另一方面，Flink CDC是基于Binlog的CDC，它根据数据库的事务日志来获取数据更改。
在出现第一个job同步延迟的情况下，可能的原因有多种。一种可能是资源竞争，例如任务的资源不足或者任务之间的资源冲突等。另一种可能是由于 Flink CDC 的任务在运行过程中遇到了异常情况导致任务失败。如果任务失败，可以尝试重新启动任务。此外，需要注意的是，Flink CDC会根据所依赖的数据库的更新频率和数据量来进行全量或增量同步，因此当第二个job提交后，对资源的需求量增大可能导致第一个job的同步速度下降。
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/575263
问题二：Flink CDC如何只取里面的INSERT和UPDATE操作，或者这么可以不回撤？
Flink CDC用FlinkSQL 接Canal Json 如何只取里面的INSERT和UPDATE操作，或者这么可以不回撤？因为我sink是ES，当回撤的时候，会先通过主键删除，导致再写入的时候，数据不完整了，如何可以避免这种通过主键删除的问题？
参考回答：
Flink CDC可以通过Flink SQL来处理Canal Json数据，只取其中的INSERT和UPDATE操作。具体来说，可以使用Flink SQL的DML语句来实现这个功能。
例如，可以使用以下语句来只获取INSERT和UPDATE操作：
SELECT * FROM canal_json WHERE operation = 'INSERT' OR operation = 'UPDATE';
此外，为了避免回撤时通过主键删除导致数据不完整的问题，可以考虑使用UPSERT（Update or Insert）操作来更新ES中的数据。这样，在更新数据时，如果数据已经存在，则更新它；否则，插入新数据。这可以避免因为先删除再插入而导致的数据丢失问题。
具体来说，可以使用Elasticsearch的API来实现UPSERT操作。例如，可以使用以下代码来更新或插入一条文档：
// 创建索引请求对象 IndexRequest indexRequest = new IndexRequest("index_name"); indexRequest.id("document_id"); // 创建文档对象 Map<String, Object> document = new HashMap<>(); document.put("field1", "value1"); document.put("field2", "value2"); // ... // 将文档对象转换为JSON字符串 String jsonString = JSON.toJSONString(document); // 创建更新请求对象 UpdateRequest updateRequest = new UpdateRequest("index_name", "document_id"); updateRequest.doc(jsonString, XContentType.JSON); // 执行更新或插入操作 RestHighLevelClient client = new RestHighLevelClient(/*...*/); client.index(updateRequest, RequestOptions.DEFAULT);
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/575262
问题三：Flink CDC这块是view,被优化拆分两部分，有没有办法共用一个view？
Flink CDC这块是view,被优化拆分两部分，有没有办法共用一个view？ 
参考回答：
Flink CDC 是一种可以捕获数据库变更的技术，其核心思想是监测并捕获数据库的变动，包括数据或数据表的插入、更新和删除等操作。然而，关于你所提问的是否有办法共用一个view，这个问题的答案可能取决于具体的使用场景和需求。在一般情况下，每个 Flink CDC 任务都会对应一个特定的 view，因为它们可能需要处理不同的数据变更。然而，如果你能确保两个任务之间的数据变更不会相互影响，那么理论上也是可以考虑共用一个 view 的。这就需要你根据实际情况来权衡利弊了。
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/575261
问题四：Flink CDC这种是跑完了自动结束任务吗，每天去提交一个任务？
Flink CDC这种是跑完了自动结束任务吗，每天去提交一个任务？
参考回答：
Flink CDC任务通常是以长期运行的方式执行的，它会持续地监听和抽取源数据库的增量更新。因此，一般情况下不需要每天手动提交一个新的任务。
您可以将CDC任务部署在Flink集群上，并通过指定的参数（如start_time或update_time）来设置任务的起始时间或增量更新的时间范围。任务会在启动后自动开始读取并处理源数据库的增量数据，并持续运行直到任务被停止或出现故障。
如果您想要以每天一次的方式运行CDC任务，您可以考虑以下几种方法：
- 使用调度工具：例如 cron、Airflow 等调度工具可以帮助您在特定的时间点触发任务的启动和停止。您可以在调度工具中配置每天执行一次CDC任务的计划，并在任务执行结束后自动停止任务。
- 结合批处理：如果您希望每天运行一次CDC任务并且只处理当天的增量数据，您可以结合使用Flink的批处理功能。您可以编写一个脚本或工作流，在每天的固定时间点启动一个Flink批处理作业来执行CDC任务，然后在作业完成后自动停止。
无论您选择哪种方式，都需要确保CDC任务能够正常连接到源数据库，并具备适当的容错机制和错误处理逻辑，以应对可能出现的故障情况。此外，还应考虑监控任务的运行状态和性能，以便及时发现和解决问题。
关于本问题的更多回答可点击原文查看：
https://developer.aliyun.com/ask/575260
问题五：Flink CDC可以实现批量抽取吗？
Flink CDC可以实现批量抽取吗？我们现在想把所有的同步任务都是用CDC，针对于实时性不高的，想通过一天一增量同步？
参考回答：
Flink CDC可以实现批量抽取。在Flink CDC中，可以通过设置checkpoint间隔和并行度来实现批量抽取。
具体来说，可以将checkpoint间隔设置为一天的时间间隔，这样每个checkpoint之间就会有一个增量数据。同时，可以设置并行度来控制每次抽取的数据量大小。通过这种方式，可以实现每天一增量同步的需求。
需要注意的是，批量抽取可能会导致数据延迟，因为需要等待checkpoint间隔才能进行下一次抽取。如果实时性要求较高，可以考虑使用流式抽取的方式。
关于本问题的更多回答可点击原文查看：
