如何从包含流 api 上的 SQL 和 NoSQL 信息的表构建聚合？

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Stream;

import static java.util.stream.Collectors.groupingBy;

public class Aaa {
    public static void main(String[] args) {
        var input = Stream.of(
                new Data(1, 1, 3, "READY", 4, 1), 
                new Data(1, 1, 3, "SUCCESS", 4, 1), 
                new Data(2, 2, 3, "READY", 4, 1), 
                new Data(2, 2, 3, "SUCCESS", 4, 1), 
                new Data(3, 3, 3, "READY", 4, 1), 
                new Data(3, 3, 3, "SUCCESS", 4, 1),
                new Data(4, 1, 3, "READY", 5, 2),
                new Data(4, 1, 3, "SUCCESS", 5, 2),
                new Data(5, 2, 3, "READY", 5, 2),
                new Data(5, 2, 3, "SUCCESS", 5, 2),
                new Data(6, 3, 3, "READY", 5, 2),
                new Data(7, 3, 3, "SUCCESS", 5, 2)
        );


        Map<Integer, List<Data>> groupByTemplate = input.collect(groupingBy(Data::templateId));

        Stream<Result> resultStream = groupByTemplate.entrySet().stream().map(e -> {
            var totalRunOrder = e.getValue().stream()
                    .filter(d -> Objects.equals(d.status, "READY"))
                    .map(Data::ordr)
                    .distinct()
                    .count();

            var successRunOrder = e.getValue().stream()
                    .filter(d -> Objects.equals(d.status, "SUCCESS") && d.curNum == d.maxNum)
                    .map(Data::ordr)
                    .distinct()
                    .count();
            
            return new Result(e.getKey(), totalRunOrder, successRunOrder);
        });

        resultStream.forEach(System.out::println);
    }
    
    record Data(int actionId, int curNum, int maxNum, String status, int templateId, int ordr){}
    record Result(int TemplateId, long totalRunOrd, long successRunOrder){}
}

在控制台中我们得到：

• 结果[TemplateId=4，totalRunOrd=1，successRunOrder=1]
• 结果[TemplateId=5，totalRunOrd=1，successRunOrder=1]

其余的可以类推。

首先，您需要在 field data+ field中创建一些与 JSON 相对应的 POJO/DTO 类orderId：

@Data
class RowEntry {
    private int orderId;
    private MyData data;
}

@Data
class MyData {
    private int actionId;
    private int curNum;
    private int maxNum;
    private int templateId;
    private String status;  // или какой-то enum
    private String errMsg;
}

从表中读取数据并以任何方便的方式（如JdbcTemplate+ ）进行转换RowMapper，输出是集合/列表RowEntry：

public static List<RowEntry> readData() {
    // ...
}

在第一遍中，构建了Map<Integer, List<RowEntry>>带有键的映射templateId，在第二遍中，计算每个列表的相应统计信息RowEntry：

public static List<Report> report(List<RowEntry> rows) {
    return rows.stream() // Stream<RowEntry>
        .collect(Collectors.groupingBy(
            row -> row.getData().getTemplateId()
        )) // Map<Integer, List<RowEntry>>
        .entrySet()
        .stream() // Stream<Map.Entry<Integer, List<RowEntry>>>
        .map(Report::new)
        .collect(Collectors.toList());
}

Report仍然需要定义表示订单/行列表的聚合统计数据的类。

@Data class Report {
    int templateId;
    int totalRun;
    int successRun;
    int failRun;
    List<FailInfo> failInfo;
}

@Data
@AllArgsConstructor 
class FailInfo {
    int orderId;
    String errMsg;
}

以经典方式一次性计算输入参数的统计数据会更好/更高效：

public Report(Map.Entry<Integer, List<RowEntry>> data) {
    this.templateId = data.getKey();

    // группировка по номеру ордера
    Map<Integer, List<RowEntry>> mapStats = data.getValue() // List<RowEntry>
        .stream()
        .collect(Collectors.groupingBy(
            RowEntry::getOrderId
        ));
    totalRun = 0;
    successRun = 0;
    failRun = 0;
    failInfo = null;
    for (Map.Entry<Integer, List<RowEntry>> byOrder : mapStats.entrySet()) {
        boolean hasReady = false;
        boolean hasSuccess = false;
        boolean hasFail = false;
        List<FailInfo> fails = new ArrayList<>();

        for (RowEntry re : byOrder.values()) {
            String status = re.getData().getStatus();
            if ("READY".equals(status)) {
                hasReady = true;
            } else if ("SUCCESS".equals(status) && re.getData().getCurNum() == re.getData().getMaxNum()) {
                hasSuccess = true;
            } else if ("FAIL".equals(status)) {
                hasFail = true;
                fails.add(new FailInfo(byOrder.getKey(), re.getData().getErrMsg()));
            }
        }
        if (hasReady) totalRun++;
        if (hasSuccess) successRun++;
        if (hasFail) {
            failRun++;
            if (!fails.isEmpty()) {
                if (failInfo == null) {
                    failInfo = new ArrayList<>(fails);
                } else {
                    failInfo.addAll(fails);
                }
            }
        }
    }
}