有没有哪位大佬知道，flink-cdc 过程中，怎么取原始数据的产生时间，比方说数据是 T日产生的，cdc程序T+1 日执行，我想取这个T日，怎么获取 ？主要是flink 本来是实时跑的，中途可能某天挂了，T+1日发现的，这种情况，也可能T+N日才修复了cdc程序，那这N日的数据，分别是怎样的的就不太容易知道了，数据是可以处理的，我想取数据当时变更的时间点，比方说A 数据 第一天新增，第二天修改，第三天删除，
cdc取到的数据recordSource里面，没看到这个时间点，看上去都是我cdc获取到数据的时间点
有没有哪位大佬知道，flink-cdc 过程中，怎么取原始数据的产生时间，比方说数据是 T日产生的？[阿里云实时计算 Flink版]
「点点赞赏，手留余香」
还没有人赞赏，快来当第一个赞赏的人吧！
在 Flink CDC 中，可以通过以下方式获取原始数据的产生时间：
使用数据库的时间戳字段
如果您的数据表中包含了时间戳字段，可以直接使用该字段获取数据的产生时间。例如，在 MySQL 数据库中，可以使用类似以下的 SQL 语句获取数据的产生时间：
sql
Copy
SELECT id, name, age, timestamp_field FROM my_table
在这个 SQL 语句中，timestamp_field 表示时间戳字段，您可以在 Flink CDC 中使用类似以下的方式来读取该字段：
java
Copy
FlinkJdbcSource source = JdbcSource
.builder()
.setDrivername(“com.mysql.jdbc.Driver”)
.setDBUrl(“jdbc:mysql://localhost:3306/test”)
.setUsername(“root”)
.setPassword(“password”)
.setQuery(“SELECT id, name, age, timestamp_field FROM my_table”)
.setRowTypeInfo(rowTypeInfo)
.build();
DataStream stream = env.addSource(source);
在这个代码中，FlinkJdbcSource 是 Flink CD
在 Flink CDC 过程中，要获取原始数据的产生时间需要考虑以下几个因素：
1. 数据源的支持：首先，要能够从数据源中获得原始数据的产生时间。不同的数据源可能提供不同的元数据信息，例如数据库日志、消息队列的时间戳等。您需要根据具体的数据源类型和特性来确定如何获取原始数据的产生时间。
2. CDC 工具和配置：Flink CDC 是一种用于捕获变更数据的工具，其提供了一种将数据源中的变更事件转换为流式数据的机制。在使用 CDC 工具时，您可以参考相应的文档和配置选项，以查看是否有与数据产生时间相关的参数或选项。这些参数可能包括最大延迟时间、时间戳字段选择等。
3. 自定义处理：如果 CDC 工具本身没有提供直接的机制来获取原始数据的产生时间，您可以考虑在 Flink 程序中进行自定义处理。例如，在 Flink 的数据流转换操作中，您可以编写自定义函数来解析数据记录，并从中提取和记录数据产生时间。然后，您可以根据需要在后续的计算逻辑中使用这个产生时间。
需要注意的是，Flink CDC 主要关注捕获变更数据并将其转换为流式数据，而不会默认提供记录数据产生时间的机制。这是因为 CDC 工具的目标是尽可能实时地从数据源获取变更事件，并将其转换为连续的流式数据，而不是关注数据产生时间的精确度。
所以，您可能需要根据具体的业务需求和数据源特性，结合自定义处理来解决获取原始数据产生时间的问题。
那你直接批量处理不就好了 ，此回答整理自钉群“【③群】Apache Flink China社区”