如何使用Flink FileSink创建dt=yyyyMMdd/hour=HH两层分桶存储目录

方案1：直接复用现有DateTimeBucketAssigner（最简实现） #

DateTimeBucketAssigner的时间格式参数支持直接插入路径分隔符/，生成的桶ID会自动携带层级结构，Flink写入时会自动创建对应多级目录，直接修改原有配置即可：

FileSink<RowData> orcSink = FileSink
        .forBulkFormat(new Path("s3a://mybucket/flink_file_sink_orc_test"), factory)
        // 格式串中加入路径分隔符直接生成两级目录
        .withBucketAssigner(new DateTimeBucketAssigner<>("'dt='yyyyMMdd/'hour='HH", ZoneId.of("Asia/Shanghai")))
        .build();

该方案无需额外开发，最终生成的路径完全符合dt=20210926/hour=13/的要求。

方案2：自定义BucketAssigner（适配复杂分桶逻辑） #

如果后续需要扩展其他分桶维度、或者需要基于事件时间做更灵活的分桶规则，可以自定义实现BucketAssigner接口：

public class CustomDtHourBucketAssigner implements BucketAssigner<RowData, String> {
    private static final long serialVersionUID = 1L;
    private final DateTimeFormatter dtFormatter;
    private final DateTimeFormatter hourFormatter;

    public CustomDtHourBucketAssigner(ZoneId zoneId) {
        this.dtFormatter = DateTimeFormatter.ofPattern("'dt='yyyyMMdd").withZone(zoneId);
        this.hourFormatter = DateTimeFormatter.ofPattern("'hour='HH").withZone(zoneId);
    }

    @Override
    public String getBucketId(RowData element, Context context) {
        // 如需使用事件时间分桶，将currentProcessingTime替换为context.timestamp()即可
        Instant processTime = Instant.ofEpochMilli(context.currentProcessingTime());
        return dtFormatter.format(processTime) + "/" + hourFormatter.format(processTime);
    }

    @Override
    public SimpleVersionedSerializer<String> getSerializer() {
        return SimpleVersionedStringSerializer.INSTANCE;
    }
}

使用时替换原有分桶器配置即可：

.withBucketAssigner(new CustomDtHourBucketAssigner(ZoneId.of("Asia/Shanghai")))

注意事项 #

• 时间格式串中的固定文本（如dt=、hour=）必须用单引号包裹，避免被DateTimeFormatter识别为时间占位符导致解析错误
• 时区配置要和业务统计时区保持一致，避免跨零点时日期、小时计算不符合预期
• 如果需要基于事件时间分桶，需要先为数据流配置正确的水位线生成策略，再在getBucketId方法中读取事件时间戳

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