如何使用带Checkpoint的Flink消费S3桶文件实现故障恢复

你当前的代码基于Flink批处理API（ExecutionEnvironment），批处理作业的容错逻辑是全量重跑——如果作业中途失败重启，整个文件会被重新处理一遍，无法做到仅处理未完成的行。要实现重启后每行仅处理一次，需要切换到流处理模式，结合Flink的Checkpoint机制和流文件源来实现。

核心思路 #

• 切换到StreamExecutionEnvironment：流处理环境支持Checkpoint，能持久化作业的读取位置和状态。
• 使用Flink官方推荐的FileSource（Flink 1.11+）：替代旧的readTextFile，它能追踪文件的读取偏移量，Checkpoint会自动保存这些偏移，重启后从上次中断的位置继续读取。
• 开启Checkpoint并配置Exactly-Once模式：确保作业状态和外部系统操作的一致性。
• 优化API调用逻辑：复用HTTP客户端（避免频繁创建销毁），同时保证API调用的幂等性（比如给每行数据加唯一标识，API端去重）。

修改后的代码示例 #

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.WatermarkStrategy;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

public class FlinkS3FileExactlyOnce {

    public static void main(String[] args) throws Exception {
        // 初始化流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启Checkpoint，配置Exactly-Once语义
        env.enableCheckpointing(30000); // 每30秒触发一次Checkpoint
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000); // 两次Checkpoint间隔至少10秒
        env.getCheckpointConfig().setCheckpointTimeout(60000); // Checkpoint超时时间

        // 配置S3文件源，读取指定文件（若需监控新增文件，可改用continuous模式）
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                new TextLineInputFormat(),
                new Path("s3://my-bucket/my-file.txt")
        ).build();

        // 从文件源获取数据流
        DataStream<String> textStream = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "S3 File Source");

        // 处理每行数据并发送到API，使用RichMapFunction复用HTTP客户端
        textStream.map(new APISender()).name("Send to API");

        // 执行作业
        env.execute("S3 File Exactly-Once Processing");
    }

    private static class APISender extends RichMapFunction<String, Void> {
        private transient CloseableHttpClient httpClient;
        private transient ValueState<Boolean> processedState;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 初始化HTTP客户端，在open方法中创建一次，复用
            httpClient = HttpClients.createDefault();
            // 状态用于记录当前行是否已处理，配合Checkpoint实现幂等
            ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>("processed", Boolean.class);
            processedState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public Void map(String line) throws Exception {
            // 幂等校验：若已处理过则跳过
            if (processedState.value() != null && processedState.value()) {
                return null;
            }

            HttpPost httpPost = new HttpPost("https://my-api-endpoint.com/api/file");
            // 构造请求体，加入唯一标识供API端去重
            String requestBody = String.format("{\"filename\": \"my-file.txt\", \"content\": \"%s\", \"unique_id\": \"%s\"}",
                    line, String.valueOf(line.hashCode()));
            httpPost.setEntity(new StringEntity(requestBody));

            try (CloseableHttpResponse response = httpClient.execute(httpPost)) {
                if (response.getStatusLine().getStatusCode() != 200) {
                    throw new Exception("API请求失败，状态码: " + response.getStatusLine().getStatusCode());
                }
                // 标记当前行已处理，状态随Checkpoint持久化
                processedState.update(true);
            }

            return null;
        }

        @Override
        public void close() throws Exception {
            // 关闭HTTP客户端
            if (httpClient != null) {
                httpClient.close();
            }
        }
    }
}

关键说明 #

• Checkpoint配置：开启后，Flink会定期保存作业的文件读取偏移和状态数据，重启后从最近的Checkpoint恢复，避免重复处理。
• FileSource优势：原生支持文件偏移追踪，无需手动维护读取位置，Checkpoint会自动管理这些状态。
• 幂等性保障：通过unique_id字段让API端实现去重，即使出现重试场景，也能保证每行数据仅生效一次。
• S3访问配置：需确保Flink集群已配置S3访问凭证（如fs.s3a.access.key、fs.s3a.secret.key等参数），保证正常访问目标桶。

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