如何配置Flink从指定时间戳开始运行任务？

完全支持，针对你描述的场景（Flink任务失败后用批处理兜底，再从指定时间点重启流任务），可以通过以下几种方式实现：

1. 针对有状态作业：重置状态+指定数据源起始时间 #

如果你的Flink任务包含状态（比如窗口聚合、状态存储），需要先处理状态再重启：

• 先停止当前失败的作业，若需要保留后续状态，先触发一个Savepoint（Flink Web UI或CLI都能操作）；若不需要之前的状态，可直接停止作业。
• 重新提交作业时，使用--fromSavepoint <savepoint路径>加载状态，同时加上--allowNonRestoredState参数（忽略和9:15之前数据相关的无用状态）。
• 在作业代码或启动参数中，配置数据源从11点对应的时间戳开始消费数据。

2. 针对无状态作业：直接配置数据源起始位置 #

如果任务没有状态，操作更简单：

• 直接修改数据源的起始消费配置，指定从目标时间戳开始读取数据即可。不同数据源的配置方式不同：
  • Kafka数据源：在FlinkKafkaConsumer中调用setStartFromTimestamp(目标时间戳)，Flink会自动定位到该时间戳之后的消息开始消费。
  • 文件/对象存储数据源：可以通过过滤条件，只处理修改时间晚于目标时间戳的文件，或者指定扫描的起始时间范围。

你的场景具体操作步骤 #

1. 确认批处理任务已经完成9-11点的数据处理，目标数据库中这部分数据已写入。
2. 停止原失败的Flink作业，若需要保留后续状态则触发Savepoint，否则直接停止。
3. 重新提交Flink作业：
   • 若有状态：加载Savepoint并忽略不需要的旧状态，同时配置数据源从11点时间戳开始消费。
   • 若无状态：直接配置数据源起始时间戳为11点，启动作业即可。

启动命令示例（Flink CLI） #

./bin/flink run -d \
  --fromSavepoint /hdfs/path/to/savepoint \
  --allowNonRestoredState \
  -c com.your.company.YourStreamingJob \
  your-job.jar \
  --start-timestamp 1620675600000  # 替换为11点对应的毫秒级时间戳

注：--start-timestamp是自定义的作业参数，需要在你的代码中读取并传递给数据源设置起始位置。

内容的提问来源于stack exchange，提问作者HEMANT PATEL