PostgreSQL中按customerId分组生成JSON对象数组的实现（超7000万行）

嘿，针对你这个7000万行交易数据的问题，我分两步给你捋清楚解决方案——既要搞定分组聚合生成JSON数组，又要保证大数据量下的性能，毕竟数据量这么大，常规操作容易卡壳。

第一步：按customerId分组聚合交易为JSON数组 #

不同的数据库/计算引擎处理JSON聚合的语法略有不同，我给你列几个常用场景的实现：

PostgreSQL #

用json_agg和json_build_object就能直接生成符合要求的JSON数组，记得按day排序，方便后续计算百分比变化：

SELECT 
  customerId,
  json_agg(
    json_build_object(
      'transactionVal', transactionVal,
      'day', day
    ) ORDER BY day
  ) AS transactions
FROM your_transaction_table
GROUP BY customerId;

MySQL 5.7+ #

用JSON_ARRAYAGG和JSON_OBJECT组合：

SELECT 
  customerId,
  JSON_ARRAYAGG(
    JSON_OBJECT(
      'transactionVal', transactionVal,
      'day', day
    ) ORDER BY day
  ) AS transactions
FROM your_transaction_table
GROUP BY customerId;

Spark SQL（最适合7000万级大数据量） #

Spark的分布式特性处理超大规模数据更高效，用collect_list聚合结构体，需要JSON字符串的话再加to_json：

-- 直接聚合为结构体数组
SELECT 
  customerId,
  collect_list(struct(transactionVal, day)) AS transactions
FROM your_transaction_table
GROUP BY customerId;

-- 转成JSON字符串格式
SELECT 
  customerId,
  to_json(collect_list(struct(transactionVal, day))) AS transactions
FROM your_transaction_table
GROUP BY customerId;

第二步：高效计算transactionVal的百分比变化 #

划重点：7000万行数据绝对不要先聚合再遍历数组计算——这种方式会把每个用户的所有交易都加载到内存，极易出现OOM或者性能雪崩。最优方案是先在原始数据上用窗口函数计算，再按需处理。

用窗口函数直接计算（推荐） #

利用LAG窗口函数获取每个用户上一次的交易值，直接计算百分比变化，性能拉满：

-- 以Spark SQL为例，其他引擎语法类似
WITH ranked_transactions AS (
  SELECT 
    customerId,
    transactionVal,
    day,
    -- 按用户分组、日期排序，获取上一条交易的金额
    LAG(transactionVal) OVER (PARTITION BY customerId ORDER BY day) AS prev_transactionVal
  FROM your_transaction_table
)
SELECT 
  customerId,
  day,
  transactionVal,
  CASE 
    WHEN prev_transactionVal IS NOT NULL 
    THEN ROUND((transactionVal - prev_transactionVal)/prev_transactionVal * 100, 2) 
    ELSE NULL -- 第一条交易没有前置数据，百分比变化设为NULL
  END AS percentage_change
FROM ranked_transactions
ORDER BY customerId, day;

如果必须先聚合为数组再计算 #

要是业务场景要求必须把百分比变化包含在聚合后的JSON数组里，Spark可以用高阶函数transform处理数组（注意：仅适合用户交易记录数不多的场景）：

WITH aggregated AS (
  SELECT 
    customerId,
    collect_list(struct(transactionVal, day)) AS transactions
  FROM your_transaction_table
  GROUP BY customerId
)
SELECT 
  customerId,
  transform(
    transactions,
    (t, idx) -> struct(
      t.transactionVal,
      t.day,
      CASE 
        WHEN idx > 0 
        THEN ROUND((t.transactionVal - transactions[idx-1].transactionVal)/transactions[idx-1].transactionVal * 100, 2)
        ELSE NULL
      END AS percentage_change
    )
  ) AS transactions_with_change
FROM aggregated;

性能优化小贴士 #

• 索引/分区优化：关系型数据库给customerId和day加联合索引；Spark中按customerId分桶或分区，减少shuffle开销。
• 数据过滤：如果不需要全量历史数据，先过滤时间范围，缩小计算数据集。
• 分布式引擎优先：7000万行数据用Spark、Flink这类分布式计算引擎，比单节点数据库效率高几个量级。

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