如何拆分嵌套JSON至关联临时表并生成Excel汇总数据？

以下提供两种主流实现方案，可根据你的技术栈和场景选择：

方案一：Python + Pandas + SQLAlchemy（兼顾拆表与Excel生成） #

该方案适合需要快速处理数据并直接生成Excel的场景，兼容绝大多数数据库。

假设前提 #

原calculations表包含主键calc_id、嵌套JSON列json_column及其他业务字段；JSON结构示例：

{
  "calc_name": "热力计算",
  "calc_date": "2024-05-20",
  "locations": [
    {"loc_id": "LOC001", "loc_name": "车间A", "counters": [{"counter_id": "CNT001", "value": 120, "unit": "℃"}]},
    {"loc_id": "LOC002", "loc_name": "车间B", "counters": [{"counter_id": "CNT003", "value": 90, "unit": "℃"}]}
  ]
}

步骤1：连接数据库并读取原始数据 #

import pandas as pd
from sqlalchemy import create_engine
import json

# 替换为你的数据库连接字符串（以PostgreSQL为例，其他数据库可调整格式）
engine = create_engine('postgresql://username:password@host:port/dbname')

# 读取原表核心字段
df_calc = pd.read_sql("SELECT calc_id, json_column, other_original_fields FROM calculations", engine)

步骤2：拆分JSON生成三个临时表的DataFrame #

1. 计算通用信息表（temp_calc_general）

提取JSON顶层通用字段，关联原表主键：

# 解析JSON列
df_calc['json_parsed'] = df_calc['json_column'].apply(json.loads)

# 生成通用信息表
df_general = df_calc[['calc_id']].copy()
df_general['calc_name'] = df_calc['json_parsed'].apply(lambda x: x.get('calc_name'))
df_general['calc_date'] = df_calc['json_parsed'].apply(lambda x: x.get('calc_date'))
# 可根据实际JSON结构添加更多顶层字段

2. 计算位置表（temp_calc_locations）

展开JSON中的locations数组，关联原表主键并生成位置自增ID：

# 展开locations数组并提取字段
df_locations = df_calc.explode('json_parsed').apply(lambda x: pd.Series(x['json_parsed']['locations']), axis=1)
# 关联原表calc_id
df_locations['calc_id'] = df_calc['calc_id'].repeat(df_calc['json_parsed'].apply(lambda x: len(x['locations'])))
# 生成位置主键
df_locations['loc_pk'] = df_locations.index + 1
df_locations = df_locations.reset_index(drop=True)

3. 位置计数器表（temp_loc_counters）

展开每个位置下的counters数组，关联位置表主键：

# 关联位置主键与计数器数组
df_loc_with_pk = df_locations[['loc_pk', 'counters']].copy()
# 展开计数器数组并提取字段
df_counters = df_loc_with_pk.explode('counters').apply(lambda x: pd.Series(x['counters']), axis=1)
# 关联位置主键
df_counters['loc_pk'] = df_loc_with_pk['loc_pk'].repeat(df_loc_with_pk['counters'].apply(lambda x: len(x)))
df_counters = df_counters.reset_index(drop=True)

步骤3：写入数据库临时表并添加关联约束 #

# 将DataFrame写入数据库临时表
df_general.to_sql('temp_calc_general', engine, if_exists='replace', index=False)
df_locations.to_sql('temp_calc_locations', engine, if_exists='replace', index=False)
df_counters.to_sql('temp_loc_counters', engine, if_exists='replace', index=False)

# 添加外键约束（可选，临时表会话结束后自动销毁）
with engine.connect() as conn:
    conn.execute("ALTER TABLE temp_calc_locations ADD CONSTRAINT fk_loc_calc FOREIGN KEY (calc_id) REFERENCES temp_calc_general(calc_id);")
    conn.execute("ALTER TABLE temp_loc_counters ADD CONSTRAINT fk_counter_loc FOREIGN KEY (loc_pk) REFERENCES temp_calc_locations(loc_pk);")
    conn.commit()

步骤4：生成包含原表字段与汇总结果的Excel #

# 生成JSON数据汇总
df_summary = df_calc[['calc_id', 'other_original_fields']].copy()
df_summary['location_count'] = df_calc['json_parsed'].apply(lambda x: len(x['locations']))
df_summary['total_counter_value'] = df_calc['json_parsed'].apply(lambda x: sum([cnt['value'] for loc in x['locations'] for cnt in loc['counters']]))

# 合并原表字段与通用信息
df_final = pd.merge(df_summary, df_general, on='calc_id', how='left')

# 写入Excel，多Sheet存储不同数据
with pd.ExcelWriter('calculation_result.xlsx') as writer:
    df_calc.to_excel(writer, sheet_name='原表数据', index=False)
    df_final.to_excel(writer, sheet_name='汇总结果', index=False)
    df_general.to_excel(writer, sheet_name='计算通用信息表', index=False)
    df_locations.to_excel(writer, sheet_name='计算位置表', index=False)
    df_counters.to_excel(writer, sheet_name='位置计数器表', index=False)

方案二：纯SQL实现（以PostgreSQL为例） #

适合仅需拆分临时表，后续通过数据库工具导出Excel的场景，利用PostgreSQL原生JSON函数处理。

步骤1：创建计算通用信息临时表 #

CREATE TEMP TABLE temp_calc_general AS
SELECT
    calc_id,
    json_column->>'calc_name' AS calc_name,
    json_column->>'calc_date' AS calc_date
    -- 按需添加其他顶层JSON字段
FROM calculations;

步骤2：创建计算位置临时表 #

CREATE TEMP TABLE temp_calc_locations AS
SELECT
    c.calc_id,
    row_number() OVER () AS loc_pk,
    loc->>'loc_id' AS loc_id,
    loc->>'loc_name' AS loc_name
FROM calculations c, json_array_elements(c.json_column->'locations') AS loc;

-- 添加外键关联通用信息表
ALTER TABLE temp_calc_locations ADD CONSTRAINT fk_loc_calc FOREIGN KEY (calc_id) REFERENCES temp_calc_general(calc_id);

步骤3：创建位置计数器临时表 #

CREATE TEMP TABLE temp_loc_counters AS
SELECT
    l.loc_pk,
    cnt->>'counter_id' AS counter_id,
    (cnt->>'value')::numeric AS value,
    cnt->>'unit' AS unit
FROM temp_calc_locations l
JOIN calculations c ON c.calc_id = l.calc_id
JOIN json_array_elements(c.json_column->'locations') AS loc ON loc->>'loc_id' = l.loc_id
JOIN json_array_elements(loc->'counters') AS cnt ON true;

-- 添加外键关联位置表
ALTER TABLE temp_loc_counters ADD CONSTRAINT fk_counter_loc FOREIGN KEY (loc_pk) REFERENCES temp_calc_locations(loc_pk);

步骤4：生成汇总查询并导出Excel #

执行以下汇总SQL，通过数据库工具（如pgAdmin）将结果导出为Excel：

SELECT
    c.calc_id,
    c.other_original_fields,
    g.calc_name,
    g.calc_date,
    COUNT(DISTINCT l.loc_pk) AS location_count,
    SUM(cnt.value) AS total_counter_value
FROM calculations c
JOIN temp_calc_general g ON c.calc_id = g.calc_id
LEFT JOIN temp_calc_locations l ON c.calc_id = l.calc_id
LEFT JOIN temp_loc_counters cnt ON l.loc_pk = cnt.loc_pk
GROUP BY c.calc_id, c.other_original_fields, g.calc_name, g.calc_date;

注意事项 #

• 需根据实际JSON结构调整字段提取逻辑，处理缺失值（如用COALESCE或x.get('field', default)）
• 临时表生命周期：PostgreSQL临时表在会话结束后自动销毁，MySQL临时表在连接关闭后销毁
• Excel生成时需注意数据类型对齐（如日期、数值格式）

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