一、前言：时间的困惑

1.1 一个让人崩溃的问题

这个问题会导致：

1.2 本文要解决的问题

• 什么是事件时间、处理时间？
• Watermark 到底是什么？
• 数据迟到怎么办？
• 各种窗口怎么选？

二、三种时间语义

2.1 时间语义图解

2.2 三种时间详解

2.3 为什么 Event Time 最重要？

三、Watermark：流处理的时钟

3.1 为什么需要 Watermark？

问题：数据是乱序的，Flink 怎么知道某个时间窗口的数据已经全部到达？

答案：Watermark！

3.2 Watermark 定义

3.3 Watermark 生成策略

代码示例：

// ========== 方式1：固定延迟（最常用）==========
DataStream<Event> stream = source
    .assignTimestampsAndWatermarks(
        WatermarkStrategy
            .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))  // 5秒延迟
            .withTimestampAssigner((event, timestamp) -> event.getEventTime())
    );

// ========== 方式2：单调递增 ==========
DataStream<Event> stream = source
    .assignTimestampsAndWatermarks(
        WatermarkStrategy
            .<Event>forMonotonousTimestamps()
            .withTimestampAssigner((event, timestamp) -> event.getEventTime())
    );

// ========== 方式3：自定义 Watermark ==========
DataStream<Event> stream = source
    .assignTimestampsAndWatermarks(
        WatermarkStrategy
            .forGenerator(ctx -> new MyWatermarkGenerator())
            .withTimestampAssigner((event, timestamp) -> event.getEventTime())
    );

// 自定义 Watermark 生成器
public class MyWatermarkGenerator implements WatermarkGenerator<Event> {
    private long maxTimestamp = Long.MIN_VALUE;
    private final long maxOutOfOrderness = 5000;  // 5秒
    
    @Override
    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }
    
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp - maxOutOfOrderness - 1));
    }
}

3.4 Watermark 传播

关键规则：下游算子的 Watermark = 所有上游 Watermark 的最小值

3.5 Watermark 与窗口触发

四、窗口详解

4.1 窗口类型全景

4.2 滚动窗口 (Tumbling Window)

特点：固定大小、不重叠、无间隙

// 事件时间滚动窗口
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .sum("value");

// 处理时间滚动窗口
stream
    .keyBy(Event::getKey)
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .sum("value");

// 带偏移量（处理时区问题）
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))  // 北京时间
    .sum("value");

4.3 滑动窗口 (Sliding Window)

特点：固定大小、可重叠、一条数据可能属于多个窗口

// 滑动窗口：窗口1分钟，每30秒滑动
stream
    .keyBy(Event::getKey)
    .window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(30)))
    .sum("value");

// 注意：如果窗口大小 / 滑动间隔 = N，则每条数据属于 N 个窗口
// 上例中 60s / 30s = 2，每条数据属于 2 个窗口

4.4 会话窗口 (Session Window)

特点：基于活动间隙、窗口大小不固定

// 静态间隙
stream
    .keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.seconds(30)))
    .sum("value");

// 动态间隙
stream
    .keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withDynamicGap(event -> {
        // 根据事件动态返回间隙
        if (event.isVip()) {
            return Time.minutes(5).toMilliseconds();
        }
        return Time.minutes(1).toMilliseconds();
    }))
    .sum("value");

4.5 计数窗口 (Count Window)

// 滚动计数窗口：每100条触发
stream
    .keyBy(Event::getKey)
    .countWindow(100)
    .sum("value");

// 滑动计数窗口：窗口100条，每10条滑动
stream
    .keyBy(Event::getKey)
    .countWindow(100, 10)
    .sum("value");

4.6 窗口选型指南

五、迟到数据处理

5.1 数据迟到的原因

5.2 迟到数据处理策略

代码示例：

// 定义侧输出标签
final OutputTag<Event> lateDataTag = new OutputTag<Event>("late-data"){};

DataStream<Result> result = stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    // 允许迟到 1 分钟
    .allowedLateness(Time.minutes(1))
    // 迟到数据输出到侧流
    .sideOutputLateData(lateDataTag)
    .sum("value");

// 获取迟到数据
DataStream<Event> lateData = result.getSideOutput(lateDataTag);

// 处理迟到数据（例如：写入特殊表、告警等）
lateData.addSink(new LateDateSink());

5.3 迟到数据处理时序

5.4 最佳实践

六、窗口函数详解

6.1 窗口函数类型

6.2 增量聚合函数

// ========== ReduceFunction ==========
// 输入输出类型相同
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce((e1, e2) -> new Event(e1.getKey(), e1.getValue() + e2.getValue()));

// ========== AggregateFunction ==========
// 更灵活，输入、累加器、输出可以不同类型
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new AggregateFunction<Event, Tuple2<Long, Long>, Double>() {
        @Override
        public Tuple2<Long, Long> createAccumulator() {
            return Tuple2.of(0L, 0L);  // (sum, count)
        }
        
        @Override
        public Tuple2<Long, Long> add(Event event, Tuple2<Long, Long> acc) {
            return Tuple2.of(acc.f0 + event.getValue(), acc.f1 + 1);
        }
        
        @Override
        public Double getResult(Tuple2<Long, Long> acc) {
            return acc.f0.doubleValue() / acc.f1;  // 计算平均值
        }
        
        @Override
        public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
            return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
        }
    });

6.3 全量窗口函数

// ProcessWindowFunction：可以获取窗口元数据
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .process(new ProcessWindowFunction<Event, Result, String, TimeWindow>() {
        @Override
        public void process(String key, Context context, Iterable<Event> elements, Collector<Result> out) {
            // 获取窗口信息
            TimeWindow window = context.window();
            long windowStart = window.getStart();
            long windowEnd = window.getEnd();
            
            // 处理所有元素
            long count = 0;
            long sum = 0;
            for (Event e : elements) {
                count++;
                sum += e.getValue();
            }
            
            out.collect(new Result(key, windowStart, windowEnd, sum, count));
        }
    });

6.4 组合使用（推荐）

// 增量聚合 + ProcessWindowFunction
// 兼顾性能（增量聚合）和灵活性（获取窗口信息）
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(
        new MyAggregateFunction(),  // 增量聚合
        new MyProcessWindowFunction()  // 封装结果
    );

// AggregateFunction
public class MyAggregateFunction implements AggregateFunction<Event, Long, Long> {
    @Override
    public Long createAccumulator() { return 0L; }
    
    @Override
    public Long add(Event event, Long acc) { return acc + event.getValue(); }
    
    @Override
    public Long getResult(Long acc) { return acc; }
    
    @Override
    public Long merge(Long a, Long b) { return a + b; }
}

// ProcessWindowFunction
public class MyProcessWindowFunction extends ProcessWindowFunction<Long, Result, String, TimeWindow> {
    @Override
    public void process(String key, Context ctx, Iterable<Long> aggResults, Collector<Result> out) {
        Long sum = aggResults.iterator().next();
        TimeWindow window = ctx.window();
        out.collect(new Result(key, window.getStart(), window.getEnd(), sum));
    }
}

七、窗口触发器与驱逐器

7.1 Trigger（触发器）

// 自定义 Trigger
public class CountTrigger extends Trigger<Event, TimeWindow> {
    private final long maxCount;
    private final ReducingStateDescriptor<Long> countDesc;
    
    public CountTrigger(long maxCount) {
        this.maxCount = maxCount;
        this.countDesc = new ReducingStateDescriptor<>("count", Long::sum, Long.class);
    }
    
    @Override
    public TriggerResult onElement(Event event, long timestamp, TimeWindow window, TriggerContext ctx) {
        ReducingState<Long> count = ctx.getPartitionedState(countDesc);
        count.add(1L);
        
        if (count.get() >= maxCount) {
            count.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }
    
    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }
    
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
    }
    
    @Override
    public void clear(TimeWindow window, TriggerContext ctx) {
        ctx.getPartitionedState(countDesc).clear();
    }
}

// 使用自定义 Trigger
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .trigger(new CountTrigger(1000))  // 每1000条触发一次
    .sum("value");

7.2 Evictor（驱逐器）

// 驱逐器：在窗口函数执行前/后移除元素
stream
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .evictor(TimeEvictor.of(Time.seconds(30)))  // 只保留最近30秒
    .sum("value");

// 常用 Evictor
// TimeEvictor: 只保留指定时间范围的数据
// CountEvictor: 只保留指定数量的数据
// DeltaEvictor: 基于差值的驱逐

八、实战案例

8.1 案例：实时 UV 统计

/**
 * 实时统计每小时 UV（独立访客数）
 */
public class HourlyUVExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        DataStream<UserBehavior> source = env
            .addSource(new UserBehaviorSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((event, ts) -> event.getTimestamp())
            );
        
        // 使用 BloomFilter 去重
        source
            .keyBy(event -> "all")  // 全局统计
            .window(TumblingEventTimeWindows.of(Time.hours(1)))
            .aggregate(new UVAggregateFunction(), new UVResultWindowFunction())
            .print();
        
        env.execute("Hourly UV");
    }
    
    // 使用 Bloom Filter 近似去重
    public static class UVAggregateFunction implements AggregateFunction<UserBehavior, Tuple2<BloomFilter<Long>, Long>, Long> {
        @Override
        public Tuple2<BloomFilter<Long>, Long> createAccumulator() {
            return Tuple2.of(
                BloomFilter.create(Funnels.longFunnel(), 1000000, 0.01),
                0L
            );
        }
        
        @Override
        public Tuple2<BloomFilter<Long>, Long> add(UserBehavior event, Tuple2<BloomFilter<Long>, Long> acc) {
            if (!acc.f0.mightContain(event.getUserId())) {
                acc.f0.put(event.getUserId());
                return Tuple2.of(acc.f0, acc.f1 + 1);
            }
            return acc;
        }
        
        @Override
        public Long getResult(Tuple2<BloomFilter<Long>, Long> acc) {
            return acc.f1;
        }
        
        @Override
        public Tuple2<BloomFilter<Long>, Long> merge(Tuple2<BloomFilter<Long>, Long> a, Tuple2<BloomFilter<Long>, Long> b) {
            a.f0.putAll(b.f0);
            return Tuple2.of(a.f0, a.f1 + b.f1);
        }
    }
}

8.2 案例：订单超时告警

/**
 * 订单30分钟未支付则告警
 */
public class OrderTimeoutExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        DataStream<Order> orders = env.addSource(new OrderSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((order, ts) -> order.getCreateTime())
            );
        
        // 使用会话窗口 + 定时器
        orders
            .keyBy(Order::getOrderId)
            .process(new OrderTimeoutFunction(30 * 60 * 1000))  // 30分钟
            .print();
        
        env.execute("Order Timeout Alert");
    }
    
    public static class OrderTimeoutFunction extends KeyedProcessFunction<String, Order, String> {
        private final long timeout;
        private ValueState<Order> orderState;
        
        public OrderTimeoutFunction(long timeout) {
            this.timeout = timeout;
        }
        
        @Override
        public void open(Configuration parameters) {
            orderState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("order", Order.class)
            );
        }
        
        @Override
        public void processElement(Order order, Context ctx, Collector<String> out) throws Exception {
            if (order.getStatus().equals("CREATED")) {
                // 订单创建，设置定时器
                orderState.update(order);
                ctx.timerService().registerEventTimeTimer(order.getCreateTime() + timeout);
            } else if (order.getStatus().equals("PAID")) {
                // 订单支付，清除状态和定时器
                Order savedOrder = orderState.value();
                if (savedOrder != null) {
                    ctx.timerService().deleteEventTimeTimer(savedOrder.getCreateTime() + timeout);
                    orderState.clear();
                }
            }
        }
        
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            Order order = orderState.value();
            if (order != null) {
                out.collect("订单超时告警: " + order.getOrderId());
                orderState.clear();
            }
        }
    }
}

九、常见问题与排查

9.1 问题速查表

9.2 调试技巧

// 打印 Watermark
stream.assignTimestampsAndWatermarks(...)
    .process(new ProcessFunction<Event, Event>() {
        @Override
        public void processElement(Event event, Context ctx, Collector<Event> out) {
            System.out.println("Event: " + event + ", Watermark: " + ctx.timerService().currentWatermark());
            out.collect(event);
        }
    });

// 使用 Metrics 监控
env.getConfig().setLatencyTrackingInterval(1000);

十、总结

10.1 核心要点

10.2 一句话总结

Event Time + Watermark + Window = Flink 流处理的核心三角

理解了这三者的关系，你就理解了 Flink 流处理的精髓。