如何高效处理CK Rollup查询结果：Java中的父子层级聚合实战

1次阅读

共计 2442 个字符，预计需要花费 7 分钟才能阅读完成。

ClickHouse 的 Rollup 功能常用于生成多维度聚合数据，但返回的结果集往往是扁平化的。例如一个商品类目分析场景，原始数据可能是这样的结构：

WITH ROLLUP 
SELECT 
    category_l1,
    category_l2,
    category_l3,
    SUM(sales)
FROM sales_data
GROUP BY 
    category_l1,
    category_l2,
    category_l3

查询结果中会出现大量包含 null 值的行，表示不同层级的汇总数据。当我们需要在 Java 中重建父子层级关系时，主要面临三个挑战：

数据膨胀问题：Rollup 会产生 2^n- 1 条记录（n 为维度数），百万级原始数据可能导致内存溢出
层级识别困难：需要根据非 null 字段判断节点所处层级
聚合计算复杂：子节点数值需要与父节点保持一致性校验

通过多次 Map 遍历标记节点层级
用 Reduce 阶段合并同类项
时间复杂度：O(n^2)

// 伪代码示例
records.stream()
       .collect(Collectors.groupingBy(r -> r.getNonNullFields()));

缺点：
– 需要多次全量遍历数据
– 无法利用树形结构的局部性原理

先找出所有根节点（全 null 行）
递归查找每个节点的直接子节点
时间复杂度：O(n log n)

优点：
– 符合树形结构的自然表达
– 可以延迟计算子树聚合值

用前缀树结构存储维度路径
边插入边计算聚合值
时间复杂度：O(n)

生产推荐：综合性能最好，但实现复杂度较高

以下是基于方案 3 的完整实现（使用 Java 17）：

public class RollupTreeBuilder {
    // 树节点定义
    @Data
    @AllArgsConstructor
    static class TreeNode {
        private String dimensionValue;
        private double metricValue;
        private Map<String, TreeNode> children = new ConcurrentHashMap<>();}

    public TreeNode build(List<RollupRecord> records) {TreeNode root = new TreeNode("TOTAL", 0.0);

        records.parallelStream().forEach(record -> {List<String> path = resolvePath(record);
            updateTree(root, path, record.getValue());
        });

        return root;
    }

    private void updateTree(TreeNode node, List<String> path, double value) {
        // 原子性更新节点值
        synchronized (node) {node.setMetricValue(node.getMetricValue() + value);
        }

        if (!path.isEmpty()) {String currentDim = path.remove(0);
            node.getChildren().computeIfAbsent(currentDim, 
                k -> new TreeNode(currentDim, 0.0));
            updateTree(node.getChildren().get(currentDim), path, value);
        }
    }
}

关键设计点：
1. 使用 ConcurrentHashMap 保证线程安全
2. 同步块仅保护计数器更新
3. 采用递归下降方式构建路径

字符串池化 ：维度值使用intern() 方法

// 在 resolvePath 方法中
return Arrays.stream(record.getDimensions())
             .filter(Objects::nonNull)
             .map(String::intern)
             .toList();

分层处理：对超过 1GB 的数据，采用磁盘 -backed 缓存

Cache<String, TreeNode> diskCache = Caffeine.newBuilder()
    .maximumWeight(1_000_000_000)
    .weigher((String key, TreeNode node) -> estimateSize(node))
    .writer(new DiskBackedWriter<>())
    .build();

预计算子树聚合值
对叶子节点启用 SIMD 指令优化
使用 -XX:+UseCompactObjectHeaders 减少对象头开销

误报：使用ConcurrentHashMap 时注意 computeIfAbsent 的阻塞特性
死锁：同步块范围要最小化

解决方案：

// 改良版同步策略
node.getChildren().compute(currentDim, (k, v) -> {if (v == null) {v = new TreeNode(currentDim, 0.0);
    }
    v.addValue(value); // 内部用 AtomicDouble
    return v;
});