Flink DataStream API 算子完全指南

概述

Flink DataStream API 提供了丰富的算子（Operators）来处理流数据。算子是 Flink 流处理的核心组件，通过组合不同的算子，可以构建复杂的流处理拓扑。本文将系统地介绍 Flink 中的各类算子，包括其功能、使用场景和最佳实践。

算子分类体系

Flink 算子可以按照不同维度进行分类：

按功能分类

1. 转换算子：map、flatMap、filter、reduce 等
2. 分区算子：keyBy、rebalance、broadcast 等
3. 窗口算子：window、windowAll、timeWindow 等
4. 连接算子：connect、union、join、coGroup 等
5. 状态算子：process、aggregate、fold 等
6. 输出算子：print、addSink、writeAsText 等

按并行度特征分类

1. 一对一算子：map、filter 等（保持并行度）
2. 重分区算子：keyBy、rebalance 等（改变数据分布）
3. 全局算子：windowAll、countWindowAll 等（并行度为 1）

数据转换算子

基础转换算子

map 算子

一对一转换，每个输入元素映射为一个输出元素。

// 基础用法
DataStream<Integer> numbers = env.fromElements(1, 2, 3, 4, 5);
DataStream<Integer> squared = numbers.map(x -> x * x);
squared.print(); // 输出: 1, 4, 9, 16, 25

// 复杂对象转换
DataStream<String> jsonStream = env.fromElements(
    "{\"name\":\"Alice\",\"age\":25}",
    "{\"name\":\"Bob\",\"age\":30}"
);

DataStream<User> userStream = jsonStream.map(new MapFunction<String, User>() {
    private final ObjectMapper mapper = new ObjectMapper();
    
    @Override
    public User map(String value) throws Exception {
        return mapper.readValue(value, User.class);
    }
});

// 使用 RichMapFunction 访问运行时上下文
public class EnrichmentMapper extends RichMapFunction<Order, EnrichedOrder> {
    private transient Map<String, UserInfo> userCache;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化资源
        userCache = loadUserCache();
        
        // 访问运行时上下文
        int subtaskIndex = getRuntimeContext().getIndexOfThisSubtask();
        int parallelism = getRuntimeContext().getNumberOfParallelSubtasks();
        System.out.printf("Mapper started: subtask %d of %d%n", subtaskIndex, parallelism);
    }
    
    @Override
    public EnrichedOrder map(Order order) throws Exception {
        UserInfo userInfo = userCache.get(order.getUserId());
        return new EnrichedOrder(order, userInfo);
    }
    
    @Override
    public void close() throws Exception {
        // 清理资源
        if (userCache != null) {
            userCache.clear();
        }
    }
}

flatMap 算子

一对多转换，每个输入元素可以映射为零个、一个或多个输出元素。

// 基础用法：分词
DataStream<String> sentences = env.fromElements(
    "Hello World",
    "Flink Streaming",
    "Real Time Processing"
);

DataStream<String> words = sentences.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String sentence, Collector<String> out) {
        for (String word : sentence.split(" ")) {
            out.collect(word.toLowerCase());
        }
    }
});

// 高级用法：条件性输出
DataStream<Event> events = env.addSource(new EventSource());

DataStream<ProcessedEvent> processed = events.flatMap(
    new RichFlatMapFunction<Event, ProcessedEvent>() {
        private transient Counter validEvents;
        private transient Counter invalidEvents;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            validEvents = getRuntimeContext()
                .getMetricGroup()
                .counter("valid_events");
            invalidEvents = getRuntimeContext()
                .getMetricGroup()
                .counter("invalid_events");
        }
        
        @Override
        public void flatMap(Event event, Collector<ProcessedEvent> out) {
            if (isValid(event)) {
                validEvents.inc();
                // 可能产生多个输出
                for (ProcessedEvent pe : processEvent(event)) {
                    out.collect(pe);
                }
            } else {
                invalidEvents.inc();
                // 不产生输出
            }
        }
    }
);

// 使用 Lambda 表达式（需要类型声明）
DataStream<String> expanded = words
    .flatMap((String word, Collector<String> out) -> {
        out.collect(word);
        out.collect(word.toUpperCase());
    })
    .returns(Types.STRING);

filter 算子

过滤数据，仅保留满足条件的元素。

// 基础过滤
DataStream<Integer> numbers = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
DataStream<Integer> evenNumbers = numbers.filter(n -> n % 2 == 0);

// 复杂条件过滤
DataStream<Transaction> transactions = env.addSource(new TransactionSource());

DataStream<Transaction> suspiciousTransactions = transactions
    .filter(new FilterFunction<Transaction>() {
        @Override
        public boolean filter(Transaction tx) {
            return tx.getAmount() > 10000 
                && tx.getLocation().equals("UNKNOWN")
                && isNightTime(tx.getTimestamp());
        }
        
        private boolean isNightTime(long timestamp) {
            LocalDateTime time = Instant.ofEpochMilli(timestamp)
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
            int hour = time.getHour();
            return hour < 6 || hour > 22;
        }
    });

// 链式过滤优化
DataStream<Order> orders = env.addSource(new OrderSource());

// 不推荐：多次过滤
DataStream<Order> filtered1 = orders.filter(order -> order.getAmount() > 100);
DataStream<Order> filtered2 = filtered1.filter(order -> order.getStatus().equals("PENDING"));

// 推荐：合并过滤条件
DataStream<Order> filteredOptimized = orders
    .filter(order -> order.getAmount() > 100 && order.getStatus().equals("PENDING"));

聚合算子

reduce 算子

对分组后的数据流进行滚动聚合。

// 基础聚合
DataStream<Tuple2<String, Integer>> wordCounts = words
    .map(word -> Tuple2.of(word, 1))
    .keyBy(tuple -> tuple.f0)
    .reduce((t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1));

// 自定义 ReduceFunction
public class TransactionSumReducer implements ReduceFunction<Transaction> {
    @Override
    public Transaction reduce(Transaction t1, Transaction t2) {
        return new Transaction(
            t1.getUserId(),
            t1.getAmount() + t2.getAmount(),
            Math.max(t1.getTimestamp(), t2.getTimestamp()),
            t1.getType()
        );
    }
}

// 复杂对象聚合
DataStream<UserActivity> activities = env.addSource(new ActivitySource());

DataStream<UserStats> userStats = activities
    .keyBy(UserActivity::getUserId)
    .reduce(new ReduceFunction<UserActivity>() {
        @Override
        public UserActivity reduce(UserActivity a1, UserActivity a2) {
            UserActivity merged = new UserActivity();
            merged.setUserId(a1.getUserId());
            merged.setClickCount(a1.getClickCount() + a2.getClickCount());
            merged.setViewCount(a1.getViewCount() + a2.getViewCount());
            merged.setTotalDuration(a1.getTotalDuration() + a2.getTotalDuration());
            merged.setLastActivityTime(Math.max(a1.getLastActivityTime(), a2.getLastActivityTime()));
            return merged;
        }
    })
    .map(activity -> new UserStats(activity));

aggregate 算子

更灵活的聚合操作，支持中间聚合状态与最终结果类型不同。

// 计算平均值
public class AverageAggregate 
        implements AggregateFunction<Integer, Tuple2<Integer, Integer>, Double> {
    
    @Override
    public Tuple2<Integer, Integer> createAccumulator() {
        return Tuple2.of(0, 0); // (sum, count)
    }
    
    @Override
    public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> acc) {
        return Tuple2.of(acc.f0 + value, acc.f1 + 1);
    }
    
    @Override
    public Double getResult(Tuple2<Integer, Integer> acc) {
        return acc.f1 == 0 ? 0.0 : (double) acc.f0 / acc.f1;
    }
    
    @Override
    public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> acc1, 
                                         Tuple2<Integer, Integer> acc2) {
        return Tuple2.of(acc1.f0 + acc2.f0, acc1.f1 + acc2.f1);
    }
}

// 使用示例
DataStream<SensorReading> readings = env.addSource(new SensorSource());

DataStream<Double> avgTemperature = readings
    .keyBy(SensorReading::getSensorId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new AverageAggregate());

// 复杂聚合：Top-K
public class TopKAggregate 
        implements AggregateFunction<Event, PriorityQueue<Event>, List<Event>> {
    private final int k;
    
    public TopKAggregate(int k) {
        this.k = k;
    }
    
    @Override
    public PriorityQueue<Event> createAccumulator() {
        return new PriorityQueue<>(k, Comparator.comparingDouble(Event::getScore));
    }
    
    @Override
    public PriorityQueue<Event> add(Event event, PriorityQueue<Event> acc) {
        acc.offer(event);
        if (acc.size() > k) {
            acc.poll(); // 移除最小元素
        }
        return acc;
    }
    
    @Override
    public List<Event> getResult(PriorityQueue<Event> acc) {
        List<Event> result = new ArrayList<>(acc);
        result.sort(Comparator.comparingDouble(Event::getScore).reversed());
        return result;
    }
    
    @Override
    public PriorityQueue<Event> merge(PriorityQueue<Event> acc1, 
                                     PriorityQueue<Event> acc2) {
        PriorityQueue<Event> merged = new PriorityQueue<>(createAccumulator());
        merged.addAll(acc1);
        merged.addAll(acc2);
        while (merged.size() > k) {
            merged.poll();
        }
        return merged;
    }
}

分组算子

keyBy 算子

按照指定的 key 对数据流进行分组，是有状态操作的基础。

// 基础分组
DataStream<Tuple2<String, Integer>> data = env.fromElements(
    Tuple2.of("A", 1),
    Tuple2.of("B", 2),
    Tuple2.of("A", 3),
    Tuple2.of("B", 4)
);

// 使用字段位置
KeyedStream<Tuple2<String, Integer>, Tuple> keyed1 = data.keyBy(0);

// 使用字段名称
KeyedStream<Tuple2<String, Integer>, String> keyed2 = data.keyBy(tuple -> tuple.f0);

// 复合键
DataStream<Order> orders = env.addSource(new OrderSource());

KeyedStream<Order, Tuple2<String, String>> keyedOrders = orders
    .keyBy(order -> Tuple2.of(order.getUserId(), order.getProductId()));

// 自定义 KeySelector
public class OrderKeySelector implements KeySelector<Order, String> {
    @Override
    public String getKey(Order order) throws Exception {
        // 复杂的 key 生成逻辑
        return String.format("%s_%s_%d", 
            order.getUserId(), 
            order.getRegion(),
            order.getTimestamp() / (1000 * 60 * 60) // 按小时分组
        );
    }
}

// 使用 POJO 字段
KeyedStream<Order, String> keyedByUser = orders.keyBy(Order::getUserId);

// 注意事项：Key 必须正确实现 hashCode 和 equals
public class CompositeKey {
    private String userId;
    private String productId;
    
    // 必须重写 hashCode 和 equals
    @Override
    public int hashCode() {
        return Objects.hash(userId, productId);
    }
    
    @Override
    public boolean equals(Object obj) {
        if (this == obj) return true;
        if (obj == null || getClass() != obj.getClass()) return false;
        CompositeKey that = (CompositeKey) obj;
        return Objects.equals(userId, that.userId) && 
               Objects.equals(productId, that.productId);
    }
}

分区算子

物理分区

rebalance 算子

轮询方式重新分配数据，实现负载均衡。

// 解决数据倾斜
DataStream<Event> skewedStream = env.addSource(new SkewedSource());

// 重新平衡数据
DataStream<Event> balancedStream = skewedStream
    .rebalance()
    .map(new ExpensiveOperation());

// 实际应用场景
DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

DataStream<ProcessedRecord> processed = kafkaStream
    .rebalance()  // 打破 Kafka 分区的数据倾斜
    .flatMap(new JsonParser())
    .filter(record -> record.isValid())
    .keyBy(ProcessedRecord::getKey)
    .process(new BusinessLogic());

rescale 算子

本地轮询，只在本地 TaskManager 的 slots 之间重分配。

// rescale vs rebalance
DataStream<Integer> stream = env.generateSequence(1, 100).setParallelism(2);

// rebalance：全局轮询
DataStream<Integer> rebalanced = stream
    .rebalance()
    .map(x -> x * 2).setParallelism(4);

// rescale：本地轮询（性能更好）
DataStream<Integer> rescaled = stream
    .rescale()
    .map(x -> x * 2).setParallelism(4);

// 性能对比示例
public class PartitioningBenchmark {
    public static void testPartitioning(StreamExecutionEnvironment env) {
        DataStream<Event> source = env
            .addSource(new HighThroughputSource())
            .setParallelism(4);
        
        // 测试不同分区策略
        source.rebalance().addSink(new DiscardingSink<>()).name("rebalance");
        source.rescale().addSink(new DiscardingSink<>()).name("rescale");
        source.shuffle().addSink(new DiscardingSink<>()).name("shuffle");
        source.broadcast().addSink(new DiscardingSink<>()).name("broadcast");
    }
}

shuffle 算子

随机分配数据到下游分区。

// 随机采样
DataStream<Record> allRecords = env.addSource(new RecordSource());

DataStream<Record> sampledRecords = allRecords
    .shuffle()
    .filter(new RandomSampler(0.1)); // 10% 采样率

// 随机采样器实现
public class RandomSampler implements FilterFunction<Record> {
    private final double samplingRate;
    private final Random random = new Random();
    
    public RandomSampler(double samplingRate) {
        this.samplingRate = samplingRate;
    }
    
    @Override
    public boolean filter(Record record) {
        return random.nextDouble() < samplingRate;
    }
}

broadcast 算子

将数据广播到所有下游任务。

// 规则广播
DataStream<Rule> rules = env
    .addSource(new RuleSource())
    .broadcast();  // 广播到所有任务

DataStream<Event> events = env.addSource(new EventSource());

// 使用广播状态
MapStateDescriptor<String, Rule> ruleStateDescriptor = 
    new MapStateDescriptor<>("rules", String.class, Rule.class);

BroadcastStream<Rule> broadcastRules = rules.broadcast(ruleStateDescriptor);

DataStream<Alert> alerts = events
    .connect(broadcastRules)
    .process(new RuleEvaluator());

// 小表 JOIN 大表
DataStream<Product> products = env.addSource(new ProductSource()); // 小表
DataStream<Order> orders = env.addSource(new OrderSource());       // 大表

DataStream<EnrichedOrder> enrichedOrders = orders
    .connect(products.broadcast())
    .process(new CoProcessFunction<Order, Product, EnrichedOrder>() {
        private final Map<String, Product> productMap = new HashMap<>();
        
        @Override
        public void processElement1(Order order, Context ctx, 
                                   Collector<EnrichedOrder> out) {
            Product product = productMap.get(order.getProductId());
            if (product != null) {
                out.collect(new EnrichedOrder(order, product));
            }
        }
        
        @Override
        public void processElement2(Product product, Context ctx, 
                                   Collector<EnrichedOrder> out) {
            productMap.put(product.getId(), product);
        }
    });

逻辑分区

partitionCustom 算子

自定义分区策略。

// 自定义分区器
public class RegionPartitioner implements Partitioner<String> {
    private final Map<String, Integer> regionToPartition = new HashMap<>();
    
    public RegionPartitioner() {
        regionToPartition.put("NORTH", 0);
        regionToPartition.put("SOUTH", 1);
        regionToPartition.put("EAST", 2);
        regionToPartition.put("WEST", 3);
    }
    
    @Override
    public int partition(String region, int numPartitions) {
        Integer partition = regionToPartition.get(region);
        if (partition != null && partition < numPartitions) {
            return partition;
        }
        // 默认哈希分区
        return Math.abs(region.hashCode() % numPartitions);
    }
}

// 使用自定义分区
DataStream<Order> orders = env.addSource(new OrderSource());

DataStream<Order> partitionedOrders = orders
    .partitionCustom(new RegionPartitioner(), Order::getRegion);

// 范围分区示例
public class RangePartitioner implements Partitioner<Long> {
    private final long[] boundaries;
    
    public RangePartitioner(long[] boundaries) {
        this.boundaries = Arrays.copyOf(boundaries, boundaries.length);
        Arrays.sort(this.boundaries);
    }
    
    @Override
    public int partition(Long key, int numPartitions) {
        int index = Arrays.binarySearch(boundaries, key);
        if (index < 0) {
            index = -index - 1;
        }
        return Math.min(index, numPartitions - 1);
    }
}

窗口算子

时间窗口

滚动窗口（Tumbling Windows）

固定大小、不重叠的窗口。

// 基础滚动窗口
DataStream<Event> events = env.addSource(new EventSource());

DataStream<WindowResult> windowedStream = events
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new EventAggregator());

// 带偏移的滚动窗口（用于时区对齐）
DataStream<Transaction> transactions = env.addSource(new TransactionSource());

DataStream<DailyStats> dailyStats = transactions
    .keyBy(Transaction::getUserId)
    .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))) // UTC-8
    .process(new DailyStatsCalculator());

// 处理时间滚动窗口
DataStream<Click> clicks = env.addSource(new ClickSource());

DataStream<ClickCount> clickCounts = clicks
    .keyBy(Click::getUserId)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    .reduce(new ClickCounter());

// 自定义窗口触发器
public class EarlyFireTrigger extends Trigger<Event, TimeWindow> {
    private final long interval;
    
    @Override
    public TriggerResult onElement(Event element, long timestamp, 
                                  TimeWindow window, TriggerContext ctx) {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            // 注册早期触发
            long timer = timestamp + interval;
            if (timer < window.maxTimestamp()) {
                ctx.registerEventTimeTimer(timer);
            }
            return TriggerResult.CONTINUE;
        }
    }
    
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, 
                                    TriggerContext ctx) {
        return time == window.maxTimestamp() ? 
            TriggerResult.FIRE_AND_PURGE : TriggerResult.FIRE;
    }
}

滑动窗口（Sliding Windows）

固定大小、可重叠的窗口。

// 基础滑动窗口
DataStream<SensorReading> readings = env.addSource(new SensorSource());

// 每5秒计算过去1分钟的平均温度
DataStream<AvgTemperature> avgTemps = readings
    .keyBy(SensorReading::getSensorId)
    .window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(5)))
    .apply(new TemperatureAverager());

// 高级用法：自定义窗口函数
public class TrendDetector 
        implements WindowFunction<Price, PriceTrend, String, TimeWindow> {
    
    @Override
    public void apply(String key, TimeWindow window, 
                     Iterable<Price> prices, 
                     Collector<PriceTrend> out) {
        List<Price> priceList = new ArrayList<>();
        prices.forEach(priceList::add);
        
        if (priceList.size() < 2) {
            return;
        }
        
        // 计算趋势
        priceList.sort(Comparator.comparingLong(Price::getTimestamp));
        double firstPrice = priceList.get(0).getValue();
        double lastPrice = priceList.get(priceList.size() - 1).getValue();
        double change = (lastPrice - firstPrice) / firstPrice * 100;
        
        PriceTrend trend = new PriceTrend(
            key,
            window.getStart(),
            window.getEnd(),
            change,
            getTrendType(change)
        );
        
        out.collect(trend);
    }
    
    private TrendType getTrendType(double change) {
        if (change > 5) return TrendType.STRONG_UP;
        if (change > 1) return TrendType.UP;
        if (change < -5) return TrendType.STRONG_DOWN;
        if (change < -1) return TrendType.DOWN;
        return TrendType.STABLE;
    }
}

// 使用增量聚合函数提高性能
DataStream<Trade> trades = env.addSource(new TradeSource());

DataStream<TradeVolume> volumes = trades
    .keyBy(Trade::getSymbol)
    .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
    .aggregate(
        new AggregateFunction<Trade, VolumeAccumulator, TradeVolume>() {
            @Override
            public VolumeAccumulator createAccumulator() {
                return new VolumeAccumulator();
            }
            
            @Override
            public VolumeAccumulator add(Trade trade, VolumeAccumulator acc) {
                acc.addTrade(trade);
                return acc;
            }
            
            @Override
            public TradeVolume getResult(VolumeAccumulator acc) {
                return acc.toTradeVolume();
            }
            
            @Override
            public VolumeAccumulator merge(VolumeAccumulator a, VolumeAccumulator b) {
                return a.merge(b);
            }
        }
    );

会话窗口

会话窗口根据活动间隔动态创建窗口。

// 基础会话窗口
DataStream<UserAction> actions = env.addSource(new UserActionSource());

// 用户会话分析（30分钟无活动则结束会话）
DataStream<UserSession> sessions = actions
    .keyBy(UserAction::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .process(new SessionAnalyzer());

// 动态会话间隔
public class DynamicSessionGapExtractor 
        implements SessionWindowTimeGapExtractor<UserAction> {
    
    @Override
    public long extract(UserAction element) {
        // 根据用户类型设置不同的会话超时
        switch (element.getUserType()) {
            case PREMIUM:
                return 60 * 60 * 1000; // 1小时
            case NORMAL:
                return 30 * 60 * 1000; // 30分钟
            case GUEST:
                return 10 * 60 * 1000; // 10分钟
            default:
                return 30 * 60 * 1000;
        }
    }
}

DataStream<UserSession> dynamicSessions = actions
    .keyBy(UserAction::getUserId)
    .window(EventTimeSessionWindows.withDynamicGap(new DynamicSessionGapExtractor()))
    .process(new SessionProcessor());

// 会话窗口处理函数
public class SessionProcessor 
        extends ProcessWindowFunction<UserAction, UserSession, String, TimeWindow> {
    
    @Override
    public void process(String userId, Context context, 
                       Iterable<UserAction> actions, 
                       Collector<UserSession> out) {
        
        List<UserAction> sessionActions = new ArrayList<>();
        actions.forEach(sessionActions::add);
        
        // 按时间排序
        sessionActions.sort(Comparator.comparingLong(UserAction::getTimestamp));
        
        UserSession session = new UserSession();
        session.setUserId(userId);
        session.setStartTime(context.window().getStart());
        session.setEndTime(context.window().getEnd());
        session.setActionCount(sessionActions.size());
        session.setDuration(context.window().getEnd() - context.window().getStart());
        
        // 分析用户行为路径
        List<String> path = sessionActions.stream()
            .map(UserAction::getActionType)
            .collect(Collectors.toList());
        session.setActionPath(path);
        
        // 计算会话价值
        double sessionValue = calculateSessionValue(sessionActions);
        session.setValue(sessionValue);
        
        out.collect(session);
    }
    
    private double calculateSessionValue(List<UserAction> actions) {
        return actions.stream()
            .mapToDouble(this::getActionValue)
            .sum();
    }
    
    private double getActionValue(UserAction action) {
        switch (action.getActionType()) {
            case "purchase": return 100.0;
            case "add_to_cart": return 10.0;
            case "view": return 1.0;
            default: return 0.0;
        }
    }
}

计数窗口

基于元素数量而非时间的窗口。

// 滚动计数窗口
DataStream<Transaction> transactions = env.addSource(new TransactionSource());

// 每100笔交易计算一次
DataStream<TransactionSummary> summaries = transactions
    .keyBy(Transaction::getUserId)
    .countWindow(100)
    .aggregate(new TransactionSummaryAggregator());

// 滑动计数窗口
DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getUserId)
    .countWindow(100, 10) // 窗口大小100，滑动步长10
    .process(new FraudDetector());

// 结合触发器的计数窗口
public class CountAndTimeTrigger extends Trigger<Event, GlobalWindow> {
    private final long maxCount;
    private final long maxTime;
    private final ReducingStateDescriptor<Long> countDesc = 
        new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);
    private final ValueStateDescriptor<Long> lastTimeDesc = 
        new ValueStateDescriptor<>("lastTime", Long.class);
    
    @Override
    public TriggerResult onElement(Event element, long timestamp, 
                                  GlobalWindow window, TriggerContext ctx) {
        ReducingState<Long> count = ctx.getPartitionedState(countDesc);
        ValueState<Long> lastTime = ctx.getPartitionedState(lastTimeDesc);
        
        count.add(1L);
        
        if (lastTime.value() == null) {
            long timer = ctx.getCurrentProcessingTime() + maxTime;
            ctx.registerProcessingTimeTimer(timer);
            lastTime.update(timer);
        }
        
        if (count.get() >= maxCount) {
            count.clear();
            lastTime.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }
        
        return TriggerResult.CONTINUE;
    }
    
    @Override
    public TriggerResult onProcessingTime(long time, GlobalWindow window, 
                                         TriggerContext ctx) {
        ReducingState<Long> count = ctx.getPartitionedState(countDesc);
        ValueState<Long> lastTime = ctx.getPartitionedState(lastTimeDesc);
        
        if (time >= lastTime.value()) {
            count.clear();
            lastTime.clear();
            return TriggerResult.FIRE_AND_PURGE;
        }
        
        return TriggerResult.CONTINUE;
    }
}

自定义窗口

实现自定义的窗口分配器和窗口函数。

// 自定义窗口分配器
public class CustomWindowAssigner extends WindowAssigner<Event, TimeWindow> {
    private final long windowSize;
    
    @Override
    public Collection<TimeWindow> assignWindows(Event element, long timestamp, 
                                               WindowAssignerContext context) {
        // 基于业务逻辑分配窗口
        long startTime = getWindowStart(element, timestamp);
        long endTime = startTime + windowSize;
        return Collections.singletonList(new TimeWindow(startTime, endTime));
    }
    
    private long getWindowStart(Event element, long timestamp) {
        // 自定义窗口起始时间计算逻辑
        // 例如：基于事件类型的不同窗口对齐
        if (element.getType() == EventType.CRITICAL) {
            // 关键事件立即开启新窗口
            return timestamp;
        } else {
            // 普通事件按小时对齐
            return timestamp - (timestamp % (60 * 60 * 1000));
        }
    }
    
    @Override
    public Trigger<Event, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }
    
    @Override
    public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
        return new TimeWindow.Serializer();
    }
    
    @Override
    public boolean isEventTime() {
        return true;
    }
}

// 自定义窗口函数
public class CustomWindowFunction 
        extends ProcessWindowFunction<Event, Result, String, TimeWindow> {
    
    private transient ValueState<Pattern> patternState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Pattern> descriptor = new ValueStateDescriptor<>(
            "pattern",
            Pattern.class
        );
        patternState = getRuntimeContext().getState(descriptor);
    }
    
    @Override
    public void process(String key, Context context, 
                       Iterable<Event> events, 
                       Collector<Result> out) throws Exception {
        
        Pattern previousPattern = patternState.value();
        Pattern currentPattern = detectPattern(events);
        
        if (previousPattern != null && isSignificantChange(previousPattern, currentPattern)) {
            out.collect(new Result(
                key,
                context.window(),
                previousPattern,
                currentPattern,
                "Pattern changed significantly"
            ));
        }
        
        patternState.update(currentPattern);
    }
    
    private Pattern detectPattern(Iterable<Event> events) {
        // 复杂的模式检测逻辑
        return new Pattern();
    }
    
    private boolean isSignificantChange(Pattern p1, Pattern p2) {
        // 判断模式是否发生显著变化
        return true;
    }
}

连接算子

流连接

union 算子

合并多个相同类型的数据流。

// 基础 union
DataStream<Event> stream1 = env.addSource(new EventSource1());
DataStream<Event> stream2 = env.addSource(new EventSource2());
DataStream<Event> stream3 = env.addSource(new EventSource3());

DataStream<Event> unionStream = stream1.union(stream2, stream3);

// 实际应用：多数据源聚合
public class MultiSourceAggregation {
    public static DataStream<Order> aggregateOrders(StreamExecutionEnvironment env) {
        // 来自不同地区的订单流
        DataStream<Order> northOrders = env.addSource(new RegionOrderSource("NORTH"));
        DataStream<Order> southOrders = env.addSource(new RegionOrderSource("SOUTH"));
        DataStream<Order> eastOrders = env.addSource(new RegionOrderSource("EAST"));
        DataStream<Order> westOrders = env.addSource(new RegionOrderSource("WEST"));
        
        // 合并所有订单流
        return northOrders
            .union(southOrders, eastOrders, westOrders)
            .keyBy(Order::getOrderId)
            .process(new OrderDeduplication()); // 去重处理
    }
}

// 多格式数据统一处理
DataStream<String> jsonStream = env.addSource(new JsonSource());
DataStream<String> xmlStream = env.addSource(new XmlSource());
DataStream<String> csvStream = env.addSource(new CsvSource());

DataStream<Record> unifiedStream = jsonStream
    .map(new JsonParser())
    .union(
        xmlStream.map(new XmlParser()),
        csvStream.map(new CsvParser())
    )
    .filter(Record::isValid)
    .keyBy(Record::getId)
    .reduce(new RecordMerger());

connect 算子

连接两个可能不同类型的数据流。

// 基础 connect
DataStream<Order> orders = env.addSource(new OrderSource());
DataStream<Payment> payments = env.addSource(new PaymentSource());

ConnectedStreams<Order, Payment> connected = orders.connect(payments);

// 使用 CoMapFunction
DataStream<String> processedStream = connected
    .map(new CoMapFunction<Order, Payment, String>() {
        @Override
        public String map1(Order order) {
            return "Order: " + order.toString();
        }
        
        @Override
        public String map2(Payment payment) {
            return "Payment: " + payment.toString();
        }
    });

// 使用 CoFlatMapFunction 实现更复杂的逻辑
public class OrderPaymentMatcher 
        extends RichCoFlatMapFunction<Order, Payment, MatchedTransaction> {
    
    private transient ValueState<Order> orderState;
    private transient ValueState<Payment> paymentState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        orderState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("order", Order.class));
        paymentState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("payment", Payment.class));
    }
    
    @Override
    public void flatMap1(Order order, Collector<MatchedTransaction> out) throws Exception {
        Payment payment = paymentState.value();
        if (payment != null && payment.getOrderId().equals(order.getOrderId())) {
            out.collect(new MatchedTransaction(order, payment));
            paymentState.clear();
        } else {
            orderState.update(order);
        }
    }
    
    @Override
    public void flatMap2(Payment payment, Collector<MatchedTransaction> out) throws Exception {
        Order order = orderState.value();
        if (order != null && order.getOrderId().equals(payment.getOrderId())) {
            out.collect(new MatchedTransaction(order, payment));
            orderState.clear();
        } else {
            paymentState.update(payment);
        }
    }
}

// 广播连接模式
DataStream<Transaction> transactions = env.addSource(new TransactionSource());
DataStream<Rule> rules = env.addSource(new RuleSource());

BroadcastStream<Rule> broadcastRules = rules.broadcast(ruleStateDescriptor);

DataStream<Alert> alerts = transactions
    .keyBy(Transaction::getUserId)
    .connect(broadcastRules)
    .process(new DynamicAlertFunction());

窗口连接

join 算子

基于窗口的内连接操作。

// 基础窗口 join
DataStream<Order> orders = env.addSource(new OrderSource());
DataStream<Shipment> shipments = env.addSource(new ShipmentSource());

DataStream<FulfilledOrder> fulfilledOrders = orders
    .join(shipments)
    .where(Order::getOrderId)
    .equalTo(Shipment::getOrderId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .apply(new JoinFunction<Order, Shipment, FulfilledOrder>() {
        @Override
        public FulfilledOrder join(Order order, Shipment shipment) {
            return new FulfilledOrder(order, shipment);
        }
    });

// 自定义 join 函数
public class OrderShipmentJoiner 
        implements JoinFunction<Order, Shipment, FulfilledOrder> {
    
    @Override
    public FulfilledOrder join(Order order, Shipment shipment) {
        FulfilledOrder fulfilled = new FulfilledOrder();
        fulfilled.setOrderId(order.getOrderId());
        fulfilled.setCustomerId(order.getCustomerId());
        fulfilled.setOrderTime(order.getTimestamp());
        fulfilled.setShipmentTime(shipment.getTimestamp());
        fulfilled.setDeliveryTime(shipment.getEstimatedDelivery());
        fulfilled.setCarrier(shipment.getCarrier());
        fulfilled.setTrackingNumber(shipment.getTrackingNumber());
        
        // 计算履行时间
        long fulfillmentTime = shipment.getTimestamp() - order.getTimestamp();
        fulfilled.setFulfillmentTime(fulfillmentTime);
        
        return fulfilled;
    }
}

// 多流 join
DataStream<Click> clicks = env.addSource(new ClickSource());
DataStream<View> views = env.addSource(new ViewSource());
DataStream<Purchase> purchases = env.addSource(new PurchaseSource());

// 先 join clicks 和 views
DataStream<ClickView> clickViews = clicks
    .join(views)
    .where(Click::getSessionId)
    .equalTo(View::getSessionId)
    .window(SlidingEventTimeWindows.of(Time.minutes(30), Time.minutes(5)))
    .apply(new ClickViewJoiner());

// 再 join 购买数据
DataStream<UserJourney> journeys = clickViews
    .join(purchases)
    .where(ClickView::getUserId)
    .equalTo(Purchase::getUserId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .apply(new UserJourneyBuilder());

coGroup 算子

类似于 join，但提供更灵活的连接逻辑。

// 基础 coGroup
DataStream<Click> clicks = env.addSource(new ClickSource());
DataStream<Impression> impressions = env.addSource(new ImpressionSource());

DataStream<CampaignStats> stats = clicks
    .coGroup(impressions)
    .where(Click::getCampaignId)
    .equalTo(Impression::getCampaignId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .apply(new CoGroupFunction<Click, Impression, CampaignStats>() {
        @Override
        public void coGroup(Iterable<Click> clicks, 
                           Iterable<Impression> impressions, 
                           Collector<CampaignStats> out) {
            
            int clickCount = 0;
            int impressionCount = 0;
            String campaignId = null;
            
            for (Click click : clicks) {
                clickCount++;
                if (campaignId == null) {
                    campaignId = click.getCampaignId();
                }
            }
            
            for (Impression impression : impressions) {
                impressionCount++;
                if (campaignId == null) {
                    campaignId = impression.getCampaignId();
                }
            }
            
            if (campaignId != null) {
                double ctr = impressionCount > 0 ? 
                    (double) clickCount / impressionCount : 0.0;
                    
                out.collect(new CampaignStats(
                    campaignId, 
                    impressionCount, 
                    clickCount, 
                    ctr
                ));
            }
        }
    });

// 高级 coGroup：左外连接实现
public class LeftOuterJoinFunction 
        implements CoGroupFunction<Order, Payment, OrderStatus> {
    
    @Override
    public void coGroup(Iterable<Order> orders, 
                       Iterable<Payment> payments, 
                       Collector<OrderStatus> out) {
        
        List<Payment> paymentList = new ArrayList<>();
        payments.forEach(paymentList::add);
        
        for (Order order : orders) {
            if (paymentList.isEmpty()) {
                // 没有匹配的支付记录
                out.collect(new OrderStatus(order, null, "UNPAID"));
            } else {
                // 有匹配的支付记录
                for (Payment payment : paymentList) {
                    out.collect(new OrderStatus(order, payment, "PAID"));
                }
            }
        }
    }
}

间隔连接

基于时间间隔的流连接。

// Interval Join
DataStream<Order> orders = env.addSource(new OrderSource());
DataStream<Payment> payments = env.addSource(new PaymentSource());

DataStream<MatchedTransaction> matched = orders
    .keyBy(Order::getOrderId)
    .intervalJoin(payments.keyBy(Payment::getOrderId))
    .between(Time.minutes(-5), Time.minutes(30)) // 订单后5分钟到30分钟内的支付
    .process(new ProcessJoinFunction<Order, Payment, MatchedTransaction>() {
        @Override
        public void processElement(Order order, Payment payment, 
                                  Context ctx, 
                                  Collector<MatchedTransaction> out) {
            
            // 验证支付金额
            if (Math.abs(order.getAmount() - payment.getAmount()) < 0.01) {
                out.collect(new MatchedTransaction(
                    order,
                    payment,
                    payment.getTimestamp() - order.getTimestamp(),
                    "MATCHED"
                ));
            } else {
                out.collect(new MatchedTransaction(
                    order,
                    payment,
                    payment.getTimestamp() - order.getTimestamp(),
                    "AMOUNT_MISMATCH"
                ));
            }
        }
    });

// 复杂的间隔连接：包含上下界的动态设置
public class DynamicIntervalJoin {
    public static DataStream<Result> performJoin(
            DataStream<EventA> streamA,
            DataStream<EventB> streamB) {
        
        return streamA
            .keyBy(EventA::getKey)
            .intervalJoin(streamB.keyBy(EventB::getKey))
            .between(Time.seconds(-10), Time.seconds(20))
            .lowerBoundExclusive() // 不包含下界
            .upperBoundExclusive() // 不包含上界
            .process(new ProcessJoinFunction<EventA, EventB, Result>() {
                @Override
                public void processElement(EventA a, EventB b, 
                                          Context ctx, 
                                          Collector<Result> out) {
                    
                    long timeDiff = b.getTimestamp() - a.getTimestamp();
                    
                    // 基于时间差的不同处理逻辑
                    if (timeDiff < 5000) {
                        out.collect(new Result(a, b, "IMMEDIATE"));
                    } else if (timeDiff < 15000) {
                        out.collect(new Result(a, b, "DELAYED"));
                    } else {
                        out.collect(new Result(a, b, "LATE"));
                    }
                }
            });
    }
}

处理函数

ProcessFunction 系列

最底层、最灵活的处理函数。

// 基础 ProcessFunction
public class EventProcessFunction extends ProcessFunction<Event, ProcessedEvent> {
    private transient ValueState<Long> countState;
    private transient ListState<Event> bufferState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Long> countDescriptor = 
            new ValueStateDescriptor<>("count", Long.class, 0L);
        countState = getRuntimeContext().getState(countDescriptor);
        
        ListStateDescriptor<Event> bufferDescriptor = 
            new ListStateDescriptor<>("buffer", Event.class);
        bufferState = getRuntimeContext().getListState(bufferDescriptor);
    }
    
    @Override
    public void processElement(Event event, Context ctx, 
                              Collector<ProcessedEvent> out) throws Exception {
        
        // 更新计数
        Long count = countState.value();
        count++;
        countState.update(count);
        
        // 缓存事件
        bufferState.add(event);
        
        // 注册定时器
        long timer = ctx.timestamp() + 60000; // 1分钟后
        ctx.timerService().registerEventTimeTimer(timer);
        
        // 访问时间戳和水位线
        Long timestamp = ctx.timestamp();
        Long watermark = ctx.timerService().currentWatermark();
        
        // 输出处理后的事件
        out.collect(new ProcessedEvent(event, count, timestamp, watermark));
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, 
                       Collector<ProcessedEvent> out) throws Exception {
        
        // 定时器触发时的处理逻辑
        List<Event> events = new ArrayList<>();
        bufferState.get().forEach(events::add);
        
        // 批量处理缓存的事件
        ProcessedEvent batchResult = processBatch(events);
        out.collect(batchResult);
        
        // 清理状态
        bufferState.clear();
    }
}

// KeyedProcessFunction 示例
public class SessionWindowFunction 
        extends KeyedProcessFunction<String, UserAction, UserSession> {
    
    private transient ValueState<UserSession> sessionState;
    private transient ValueState<Long> lastActionTime;
    private static final long SESSION_TIMEOUT = 30 * 60 * 1000; // 30分钟
    
    @Override
    public void open(Configuration parameters) throws Exception {
        sessionState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("session", UserSession.class));
        lastActionTime = getRuntimeContext().getState(
            new ValueStateDescriptor<>("lastAction", Long.class));
    }
    
    @Override
    public void processElement(UserAction action, Context ctx, 
                              Collector<UserSession> out) throws Exception {
        
        UserSession session = sessionState.value();
        if (session == null) {
            session = new UserSession(action.getUserId());
        }
        
        // 更新会话
        session.addAction(action);
        session.updateLastActionTime(action.getTimestamp());
        
        // 设置会话超时定时器
        Long lastTime = lastActionTime.value();
        if (lastTime != null) {
            ctx.timerService().deleteEventTimeTimer(lastTime + SESSION_TIMEOUT);
        }
        
        long newTimer = action.getTimestamp() + SESSION_TIMEOUT;
        ctx.timerService().registerEventTimeTimer(newTimer);
        
        // 更新状态
        sessionState.update(session);
        lastActionTime.update(action.getTimestamp());
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, 
                       Collector<UserSession> out) throws Exception {
        
        // 会话超时，输出会话并清理状态
        UserSession session = sessionState.value();
        if (session != null) {
            session.setEndTime(timestamp - SESSION_TIMEOUT);
            out.collect(session);
            
            sessionState.clear();
            lastActionTime.clear();
        }
    }
}

// CoProcessFunction 示例
public class OrderPaymentProcessor 
        extends CoProcessFunction<Order, Payment, Transaction> {
    
    private transient MapState<String, Order> pendingOrders;
    private transient MapState<String, Payment> pendingPayments;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        pendingOrders = getRuntimeContext().getMapState(
            new MapStateDescriptor<>("orders", String.class, Order.class));
        pendingPayments = getRuntimeContext().getMapState(
            new MapStateDescriptor<>("payments", String.class, Payment.class));
    }
    
    @Override
    public void processElement1(Order order, Context ctx, 
                               Collector<Transaction> out) throws Exception {
        
        Payment payment = pendingPayments.get(order.getOrderId());
        if (payment != null) {
            // 找到匹配的支付
            out.collect(createTransaction(order, payment));
            pendingPayments.remove(order.getOrderId());
        } else {
            // 缓存订单，等待支付
            pendingOrders.put(order.getOrderId(), order);
            
            // 设置超时定时器（30分钟）
            ctx.timerService().registerProcessingTimeTimer(
                ctx.timerService().currentProcessingTime() + 30 * 60 * 1000
            );
        }
    }
    
    @Override
    public void processElement2(Payment payment, Context ctx, 
                               Collector<Transaction> out) throws Exception {
        
        Order order = pendingOrders.get(payment.getOrderId());
        if (order != null) {
            // 找到匹配的订单
            out.collect(createTransaction(order, payment));
            pendingOrders.remove(payment.getOrderId());
        } else {
            // 缓存支付，等待订单
            pendingPayments.put(payment.getOrderId(), payment);
        }
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, 
                       Collector<Transaction> out) throws Exception {
        
        // 清理超时的待处理数据
        List<String> timeoutOrders = new ArrayList<>();
        
        for (Map.Entry<String, Order> entry : pendingOrders.entries()) {
            if (entry.getValue().getTimestamp() + 30 * 60 * 1000 < timestamp) {
                timeoutOrders.add(entry.getKey());
                
                // 输出超时交易
                out.collect(createTimeoutTransaction(entry.getValue()));
            }
        }
        
        timeoutOrders.forEach(orderId -> {
            try {
                pendingOrders.remove(orderId);
            } catch (Exception e) {
                // 处理异常
            }
        });
    }
}