技术问询：Bucketing Sink能否基于事件时间创建Bucket？

1. 是否可以基于事件时间创建Bucket？ #

绝对可以！虽然Flink的Bucketing Sink默认采用系统（处理）时间来划分Bucket，但它提供了足够的扩展性，完全支持切换为基于事件时间的Bucket分配逻辑。

2. 如何改为基于事件时间分配？ #

核心思路是自定义Bucketer接口实现类，在getBucketId方法中提取事件自身携带的时间字段来生成Bucket ID，而非依赖当前系统时间。

举个Java代码示例（假设你的事件类为Event，包含getEventTime()方法返回时间戳）：

public class CustomEventTimeBucketer<T> implements Bucketer<T> {
    // 根据需求定义时间格式，比如按小时划分Bucket
    private final SimpleDateFormat bucketFormat = new SimpleDateFormat("yyyy-MM-dd-HH");

    @Override
    public String getBucketId(T element, BucketAssigner.Context context) {
        // 转换为你的事件类型，提取事件时间
        Event event = (Event) element;
        return bucketFormat.format(new Date(event.getEventTime()));
    }
}

然后在配置Bucketing Sink时，替换默认的Bucketer：

// 初始化Sink
BucketingSink<Event> bucketingSink = new BucketingSink<>("hdfs://your/sink/path");
// 设置自定义的事件时间Bucketer
bucketingSink.setBucketer(new CustomEventTimeBucketer<>());

另外补充一点：如果使用的是Flink 1.11及以后的版本，官方更推荐使用FileSink替代旧的Bucketing Sink。FileSink中可以直接使用内置的EventTimeBucketAssigner，或者自定义BucketAssigner来实现事件时间划分，用法逻辑和上面类似，配置起来更简洁。

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