如何配置Flink StreamingFileSink基于水位线滚动文件并解决回填OOM问题

你碰到的这个问题我太熟悉了——在回溯历史数据或者从最早偏移量启动作业时，Flink处理速度快得离谱，而基于处理时间的滚动策略完全跟不上事件时间的分布，直接导致同时打开几百上千个Bucket，每个都占着压缩缓冲区内存，最后妥妥OOM。

核心破局点就是：把文件滚动的触发逻辑从处理时间绑定到事件时间（水位线），让文件关闭的时机和数据的时间进展挂钩，而不是Flink的处理速度。

问题到底出在哪？ #

你的BatchingCheckpointRollingPolicy完全靠处理时间和文件大小来触发滚动：要么文件到128M，要么Bucket创建/最后更新超时。但回溯的时候，Flink会在几分钟内把过去7天的所有分区数据都读进来，这些数据的eventTs分布在几百个小时分区里，但处理时间几乎是同一时刻——这就意味着Flink会为每个历史分区都打开一个Bucket，却因为处理时间没到超时阈值，一直攥着这些Bucket的内存不放，最终撑爆。

解决方案：基于水位线的滚动策略 #

要实现这个需求，你需要做三件事：确保作业正确用事件时间、自定义结合水位线的滚动策略、配置Sink使用新策略。

1. 先把事件时间和水位线配置对 #

这是基础中的基础，如果水位线都不准，后面的逻辑全白搭。假设你的事件类是Event，eventTs是毫秒级时间戳：

DataStream<Event> eventStream = kafkaSource
    .assignTimestampsAndWatermarks(WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofMinutes(5)) // 按业务乱序情况调，比如允许5分钟乱序
        .withTimestampAssigner((event, ignored) -> event.getEventTs())
    );

如果你的回溯数据是严格按时间递增的，可以换成forMonotonousTimestamps()，水位线推进更快。

2. 自定义滚动策略：水位线触发+大小/时间兜底 #

我们要写一个CheckpointRollingPolicy，核心逻辑是：当水位线超过当前Bucket对应时间分区的上限时，强制关闭这个Bucket的文件。而且必须在checkpoint时判断——因为只有checkpoint时，Flink能保证所有Task的水位线都对齐，此时关闭Bucket才不会丢后续数据。

假设你的BucketID格式是eventType=XXX/hour=YYYYMMDDHH，代码实现如下：

public class WatermarkTriggeredRollingPolicy<IN> extends CheckpointRollingPolicy<IN, String> {
    private final long maxSizeBytes;
    private final long maxInactiveMillis;
    private final long maxCreationMillis;

    public WatermarkTriggeredRollingPolicy(long maxSizeBytes, long maxInactiveMillis, long maxCreationMillis) {
        this.maxSizeBytes = maxSizeBytes;
        this.maxInactiveMillis = maxInactiveMillis;
        this.maxCreationMillis = maxCreationMillis;
    }

    // 保留文件大小触发逻辑，达到128M就滚动
    @Override
    public boolean shouldRollOnEvent(PartFileInfo<String> partFileState, IN element) {
        return partFileState.getSize() >= maxSizeBytes;
    }

    // 保留处理时间兜底，防止极端情况（比如某个分区只有几条数据，水位线一直没推进到分区结束）
    @Override
    public boolean shouldRollOnProcessingTime(PartFileInfo<String> partFileState, long currentTime) {
        long timeSinceCreated = currentTime - partFileState.getCreationTime();
        long timeSinceUpdated = currentTime - partFileState.getLastUpdateTime();
        return timeSinceCreated >= maxCreationMillis || timeSinceUpdated >= maxInactiveMillis;
    }

    // 核心逻辑：checkpoint时用水位线判断是否滚动
    @Override
    public boolean shouldRollOnCheckpoint(PartFileInfo<String> partFileState, long checkpointId) {
        // 从BucketID里解析出对应的小时分区
        String bucketId = partFileState.getBucketId();
        String hourPartition = bucketId.split("/hour=")[1];
        // 把分区字符串转成该小时的结束时间戳（比如2024052010 -> 2024-05-20 10:59:59.999）
        LocalDateTime hourEnd = LocalDateTime.parse(hourPartition, DateTimeFormatter.ofPattern("yyyyMMddHH"))
                .plusHours(1)
                .minusNanos(1);
        long partitionEndTs = hourEnd.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();

        // 获取当前作业的全局水位线
        long currentWatermark = getRuntimeContext().getMetricGroup().getIOMetricGroup().getWatermark();

        // 水位线超过分区结束时间，说明该分区不会再有新数据进来了，可以滚动关闭
        return currentWatermark >= partitionEndTs;
    }

    // 方便创建的静态方法
    public static <IN> WatermarkTriggeredRollingPolicy<IN> create(long maxSizeBytes, long maxInactiveMillis, long maxCreationMillis) {
        return new WatermarkTriggeredRollingPolicy<>(maxSizeBytes, maxInactiveMillis, maxCreationMillis);
    }
}

3. 把策略配置到StreamingFileSink里 #

创建Sink的时候，替换掉原来的滚动策略就行：

StreamingFileSink<Event> s3Sink = StreamingFileSink
    .forRowFormat(new Path("s3://your-bucket/base-path"), new SimpleStringEncoder<>("UTF-8"))
    .withBucketAssigner(new BucketAssigner<Event, String>() {
        @Override
        public String getBucketId(Event event, Context context) {
            // 生成你要的路径格式
            LocalDateTime eventHour = LocalDateTime.ofInstant(Instant.ofEpochMilli(event.getEventTs()), ZoneId.systemDefault());
            String hourStr = eventHour.format(DateTimeFormatter.ofPattern("yyyyMMddHH"));
            return String.format("eventType=%s/hour=%s", event.getEventType(), hourStr);
        }

        @Override
        public SimpleVersionedSerializer<String> getSerializer() {
            return SimpleVersionedStringSerializer.INSTANCE;
        }
    })
    .withRollingPolicy(WatermarkTriggeredRollingPolicy.create(
        128 * 1024 * 1024, // 128MB大小阈值
        30 * 60 * 1000,     // 30分钟无活动兜底关闭
        60 * 60 * 1000      // 1小时创建后兜底关闭
    ))
    .withBucketCheckInterval(1000) // 调整Bucket检查间隔，按需设置
    .setBucketMaxParallelism(100)  // 限制同时打开的Bucket数量，防止内存爆掉
    .build();

额外优化小技巧 #

1. 限制Bucket并发数：上面的setBucketMaxParallelism(100)一定要加，哪怕你的水位线逻辑完美，极端情况下还是可能有大量Bucket同时打开，这个参数能帮你兜底。
2. 调整压缩缓冲区：如果用Snappy压缩，可以通过Flink配置fs.s3a.block.size或者调整编码器的缓冲区大小，减少每个Bucket的内存占用。
3. 水位线调优：回溯时如果数据是有序的，直接用forMonotonousTimestamps()，水位线推进更快，Bucket关闭更及时。

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