如何在BigQuery/MySQL/SQLAlchemy中合并含JSON的两行为单行

针对同一sid关联2条独立记录、每条记录data字段存储独立JSON对象的场景，目标聚合结果为单条sid记录，data字段结构为{"1": 第一条原JSON, "2": 第二条原JSON}，以下是不同环境的可直接运行实现：

• BigQuery 环境
  利用原生JSON函数+窗口函数保证顺序稳定，若data字段为字符串存储的JSON，替换为PARSE_JSON(data)即可：

  WITH ranked AS (
    SELECT
      sid,
      data,
      ROW_NUMBER() OVER (PARTITION BY sid ORDER BY create_time ASC) AS rn -- 排序字段可按业务规则替换
    FROM your_source_table
    QUALIFY COUNT(1) OVER (PARTITION BY sid) = 2 -- 仅保留恰好2条记录的sid，避免脏数据
  )
  SELECT
    sid,
    JSON_OBJECT(
      '1', MAX(IF(rn = 1, data, NULL)),
      '2', MAX(IF(rn = 2, data, NULL))
    ) AS data
  FROM ranked
  GROUP BY sid
  

• MySQL 环境
  8.0及以上版本支持窗口函数，写法简洁：

  WITH ranked AS (
    SELECT
      sid,
      data,
      ROW_NUMBER() OVER (PARTITION BY sid ORDER BY id ASC) AS rn -- 按主键排序保证顺序固定
    FROM your_source_table
  )
  SELECT
    sid,
    JSON_OBJECT(
      '1', MAX(CASE WHEN rn = 1 THEN data END),
      '2', MAX(CASE WHEN rn = 2 THEN data END)
    ) AS data
  FROM ranked
  GROUP BY sid
  HAVING COUNT(*) = 2;
  

  5.7版本无窗口函数，可通过自关联实现：

  SELECT
    t1.sid,
    JSON_OBJECT(
      '1', MIN(IF(t1.id < t2.id, t1.data, t2.data)),
      '2', MAX(IF(t1.id < t2.id, t1.data, t2.data))
    ) AS data
  FROM your_source_table t1
  INNER JOIN your_source_table t2 
    ON t1.sid = t2.sid AND t1.id <> t2.id
  GROUP BY t1.sid
  HAVING COUNT(*) = 2;
  

• SQLAlchemy 环境
  以适配MySQL8.0/BigQuery的ORM写法为例，可直接集成到现有代码逻辑：

  from sqlalchemy import func, select, case
  from your_models import YourTable # 替换为实际业务表模型
  
  # 构造带行号的CTE子查询
  ranked_cte = (
      select(
          YourTable.sid,
          YourTable.data,
          func.row_number()
              .over(partition_by=YourTable.sid, order_by=YourTable.id.asc())
              .label("rn")
      )
      .cte("ranked_data")
  )
  
  # 构造最终聚合查询
  merge_query = (
      select(
          ranked_cte.c.sid,
          func.json_object(
              "1", func.max(case((ranked_cte.c.rn == 1, ranked_cte.c.data))),
              "2", func.max(case((ranked_cte.c.rn == 2, ranked_cte.c.data)))
          ).label("data")
      )
      .group_by(ranked_cte.c.sid)
      .having(func.count() == 2)
  )
  
  # 执行查询获取结果
  # result = session.execute(merge_query).all()
  

  若使用的数据库方言对JSON函数名有差异，替换func.json_object为对应函数即可，比如BigQuery下无需修改，SQLAlchemy会自动适配大小写。

注意：所有实现中的排序字段必须可以唯一区分同sid下的两条记录，否则会出现键1、键2对应内容随机的问题；如果后续需要合并同sid下超过2条记录，只需扩展JSON_OBJECT的键值对、匹配对应行号即可。

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