Flink按需触发Checkpoint生成Avro文件方案咨询（Flink 1.15.4）

1. 是否可替代定时间隔，按需触发生成Avro文件 #

可以实现。当前逻辑是基于Checkpoint触发Avro文件生成，核心思路为关闭定时Checkpoint，改为按需触发Checkpoint，以此间接实现按需生成Avro文件，具体操作要点：

• 关闭自动定时Checkpoint：配置文件中设置execution.checkpointing.interval: 0，或代码中调用env.disableCheckpointing()。
• 保留原有基于Checkpoint生成Avro文件的逻辑，比如通过CheckpointListener监听Checkpoint完成事件，触发Avro文件写入；或使用支持Checkpoint触发的Sink组件。
• 后续通过手动触发Checkpoint的方式，按需触发Avro文件生成。

2. 强制触发Checkpoint的实现方式 #

REST API方式 #

Flink 1.15.4支持通过REST API手动触发Checkpoint，请求格式如下：

POST /jobs/<job_id>/checkpoints

示例请求（curl调用）：

curl -X POST http://<jobmanager_address>:<port>/jobs/<your_job_id>/checkpoints \
  -H "Content-Type: application/json" \
  -d '{"mode": "EXACTLY_ONCE", "savepointType": "CHECKPOINT"}'

参数说明：

• <jobmanager_address>为JobManager的REST服务地址，<port>默认是8081。
• <your_job_id>可通过Flink UI或GET /jobs接口获取。

Java代码实现方式 #

可通过JobClient触发Checkpoint，示例代码：

import org.apache.flink.api.common.JobID;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.rest.RestClusterClient;
import org.apache.flink.configuration.Configuration;

public class ManualCheckpointTrigger {
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        config.setString("rest.address", "<jobmanager_address>");
        config.setInteger("rest.port", 8081);

        try (ClusterClient<String> client = new RestClusterClient<>(config, "default")) {
            JobID jobId = JobID.fromHexString("<your_job_id>");
            // 触发普通Checkpoint，第二个参数设为true则同时生成Savepoint
            client.triggerCheckpoint(jobId, false).get();
        }
    }
}

注意：项目需引入flink-clients依赖包。

配置方式 #

没有直接通过配置强制触发Checkpoint的方式，配置仅用于开启/关闭自动定时Checkpoint，按需触发只能通过上述API或代码方式实现。

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