使用Flink的Reduce算子时为何得到两条结果？

嘿，我帮你分析下这个问题哈！你现在遇到的情况是Flink从Kafka取数后，窗口输出了两条记录，而不是预期的单条最小frequency的结果，主要是这几个原因导致的：

1. 时间特征与窗口类型不匹配 #

你明明设置了EventTime作为任务的时间特征，却用了TumblingProcessingTimeWindows（处理时间窗口）。处理时间窗口是根据Flink任务的系统时间来切分窗口的，如果Kafka消息不是同时到达Flink，就会被分到不同的窗口里，自然会输出多条结果。而你的需求应该是基于消息本身的时间戳（也就是KafkaEvent里的timestamp）来划分窗口，这样同属一个时间窗口的消息才会被一起处理，最终输出单条最小记录。

2. Watermark生成与事件时间戳的问题 #

虽然你配置了CustomWatermarkExtractor，但如果这个类没有正确生成Watermark，或者KafkaEvent的timestamp字段在发送/解析时没被正确赋值，EventTime窗口就没法正常工作。比如如果消息的时间戳没设置，EventTime窗口就没法判断消息属于哪个窗口，结果还是会乱。

3. 代码里的小细节问题 #

你的WindowFunction里把变量名写错啦——KafkaEvent max = minReadings.iterator().next();，这里应该叫minEvent才对，虽然不影响逻辑，但容易搞混，建议修正过来。

给你改好的代码示例 #

核心修改就是把窗口换成EventTime类型，同时补全了CustomWatermarkExtractor的正确实现，还修正了变量名：

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment().setParallelism(1);
    env.getConfig().disableSysoutLogging();
    env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
    env.enableCheckpointing(5000); // 每5秒创建一个检查点
    env.getConfig().setGlobalJobParameters(parameterTool); // 让参数在Web界面可见
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    DataStream<Tuple2<String, Tuple1<Integer>>> input = env
            .addSource(
                    new FlinkKafkaConsumer010<>(
                            parameterTool.getRequired("input-topic"),
                            new KafkaEventSchema(),
                            parameterTool.getProperties())
                    // 这里补全了Watermark提取器的实现，你可以根据业务调整乱序容忍时间
                    .assignTimestampsAndWatermarks(new CustomWatermarkExtractor()))
            .keyBy("word")
            // 替换为基于事件时间的滚动窗口
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .reduce(new MyReduceFunction(), new MyWindowFunction());

    input.print() ;
    env.execute("Kafka 0.10 Example");
}

private static class MyReduceFunction implements ReduceFunction<KafkaEvent> {
    @Override
    public KafkaEvent reduce(KafkaEvent r1, KafkaEvent r2) {
        // 每次合并都保留frequency更小的事件
        return r1.getFrequency() > r2.getFrequency() ? r2 : r1;
    }
}

private static class MyWindowFunction implements WindowFunction<KafkaEvent, Tuple2<String, Tuple1<Integer>>, Tuple, TimeWindow> {
    @Override
    public void apply(Tuple key, TimeWindow window, Iterable<KafkaEvent> minReadings, Collector<Tuple2<String, Tuple1<Integer>>> out) {
        // 经过reduce后，Iterable里只会有一个最小的事件
        KafkaEvent minEvent = minReadings.iterator().next();
        out.collect(new Tuple2<>(minEvent.getWord(), new Tuple1<>(minEvent.getFrequency())));
    }
}

// 基于事件时间戳生成Watermark的示例实现，允许5秒的乱序
private static class CustomWatermarkExtractor implements AssignerWithPeriodicWatermarks<KafkaEvent> {
    private long currentMaxTimestamp = 0;
    private final long maxOutOfOrderness = 5000; // 可根据业务调整

    @Override
    public long extractTimestamp(KafkaEvent event, long previousElementTimestamp) {
        long timestamp = event.getTimestamp();
        currentMaxTimestamp = Math.max(currentMaxTimestamp, timestamp);
        return timestamp;
    }

    @Override
    public Watermark getCurrentWatermark() {
        // Watermark = 当前最大时间戳 - 允许的乱序时间，确保所有延迟在容忍范围内的消息都被处理
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

public static class KafkaEvent {
    private String word;
    private int frequency;
    private long timestamp;

    // 一定要添加getter和setter，不然Flink没法访问字段
    public String getWord() { return word; }
    public void setWord(String word) { this.word = word; }
    public int getFrequency() { return frequency; }
    public void setFrequency(int frequency) { this.frequency = frequency; }
    public long getTimestamp() { return timestamp; }
    public void setTimestamp(long timestamp) { this.timestamp = timestamp; }
}

// 确保Kafka消息的序列化/反序列化正确，这里假设用JSON格式
public static class KafkaEventSchema implements DeserializationSchema<KafkaEvent>, SerializationSchema<KafkaEvent> {
    private final ObjectMapper mapper = new ObjectMapper();

    @Override
    public KafkaEvent deserialize(byte[] message) throws IOException {
        return mapper.readValue(message, KafkaEvent.class);
    }

    @Override
    public byte[] serialize(KafkaEvent event) throws IOException {
        return mapper.writeValueAsBytes(event);
    }

    @Override
    public boolean isEndOfStream(KafkaEvent nextElement) {
        return false;
    }
}

额外要注意的点 #

1. 消息时间戳要正确设置：发送Kafka消息时，记得把KafkaEvent的timestamp字段设为消息产生的时间（比如System.currentTimeMillis()），不然EventTime窗口没法正确划分。
2. 调整乱序容忍时间：如果你的业务消息几乎没有乱序，可以把maxOutOfOrderness设为0；如果有较多延迟消息，就调大这个值。
3. 检查Kafka消费配置：确保auto.offset.reset的配置符合你的需求，避免重复消费导致重复输出。

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