Hazelcast Jet：如何实现分组后一次性获取每组数据完成聚合？

你遇到的问题根源在于用错了聚合方法：rollingAggregate是滚动聚合，设计逻辑就是数据分批到达时就更新对应分组的结果，所以同一个分组会多次触发处理。而你需要的是每个分组所有数据收集完成后仅触发一次聚合，这时候应该用BatchStage的aggregate方法——它会等分组的全量数据到齐后才执行聚合逻辑，既满足单分组仅处理一次的需求，也不会一次性加载所有分组，适配百万级数据量的性能要求。

调整后的代码示例 #

假设你的JDBC返回数据是Object[]格式（如果是自定义对象，逻辑类似）：

BatchStage<Object[]> jdbcBatchStage = // 来自JDBC的数据源

jdbcBatchStage
    // 1. 生成唯一分组键：id + country的组合（用SimpleEntry避免字符串拼接的冲突风险）
    .groupingKey(data -> {
        Integer id = (Integer) data[0];
        String country = (String) data[1];
        return new AbstractMap.SimpleEntry<>(id, country);
    })
    // 2. 用aggregate替换rollingAggregate，直接指定对amount字段求和（比先存List再手动求和高效）
    .aggregate(AggregateOperations.sum(data -> (Long) data[3]))
    // 3. 映射聚合结果为你需要的输出格式
    .map(groupResult -> {
        Map.Entry<Map.Entry<Integer, String>, Long> entry = groupResult;
        Integer id = entry.getKey().getKey();
        String country = entry.getKey().getValue();
        Long totalAmount = entry.getValue();
        // 组装结果行，id2字段无聚合意义可设为null或按需处理
        return new Object[]{id, country, null, totalAmount};
    });

如果你的数据是自定义对象（比如Record类），代码会更简洁：

// 自定义数据类
class Record {
    private Integer id;
    private String country;
    private String id2;
    private Long amount;
    // 省略getter、构造方法
}

BatchStage<Record> jdbcBatchStage = // 来自JDBC的数据源

jdbcBatchStage
    .groupingKey(record -> new AbstractMap.SimpleEntry<>(record.getId(), record.getCountry()))
    .aggregate(AggregateOperations.sum(Record::getAmount))
    .map(groupResult -> new Object[]{
        groupResult.getKey().getKey(),
        groupResult.getKey().getValue(),
        null,
        groupResult.getValue()
    });

关键说明 #

1. 为什么用aggregate？
   它会为每个分组缓存数据，直到该分组的所有数据处理完成，才一次性触发聚合计算，每个分组只会进入map方法一次，完全符合你"单分组仅处理一次"的要求。
2. 性能优势
   使用sum聚合操作无需将全量分组数据存入List再遍历求和，而是在数据到达时直接累加，内存占用更低，适合百万级数据场景。
3. 分组键的安全性
   用SimpleEntry（或自定义不可变对象）替代字符串拼接，避免id/country包含特殊字符（比如逗号）导致的分组键冲突问题。

测试结果验证 #

针对你提供的输入数据，调整后的代码会输出你期望的结果：

[['id','country','id2','amount']
 [3638, Dominican Republic, null, 1068600],  
 [3668, USA, null, 1187856]]

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