SQL Server解析可变结构JSON列及Python替代方案

一、SQL Server 解决方案 #

针对非固定结构的JSON数组，可通过多层OPENJSON解析提取目标字段：

SELECT
    t._id AS row_id,
    -- 提取created_at字段
    MAX(CASE WHEN j2.[key] = 'created_at' THEN j2.[value] END) AS created_at,
    -- 解析from对象，提取名称和ID
    MAX(CASE WHEN j2.[key] = 'from' THEN JSON_VALUE(j2.[value], '$.name') END) AS from_name,
    MAX(CASE WHEN j2.[key] = 'from' THEN JSON_VALUE(j2.[value], '$.userId') END) AS from_id,
    -- 匹配动态*-parecer键，提取对应的文本内容
    MAX(CASE WHEN j2.[key] LIKE '%-parecer' THEN JSON_VALUE(j2.[value], '$.*-texto') END) AS text
FROM
    my.table AS t
    -- 拆分JSON数组为单个对象
    CROSS APPLY OPENJSON(t.observacoes) AS j1
    -- 展开每个对象的所有键值对
    CROSS APPLY OPENJSON(j1.[value]) AS j2
GROUP BY
    t._id, j1.[key] -- 按原表ID和数组元素分组，保证每个数组对象对应一行结果
ORDER BY
    t._id, created_at;

关键逻辑： #

1. 第一层OPENJSON将JSON数组拆分为独立对象；
2. 第二层OPENJSON展开每个对象的所有键值对，兼容任意键名；
3. 通过CASE语句筛选目标字段：
   • 直接匹配created_at键获取时间；
   • 用JSON_VALUE解析from嵌套对象中的name和userId；
   • 对带-parecer后缀的动态键，用通配符路径$.*-texto提取文本；
4. 分组聚合确保每个数组元素对应一条结果。

二、Python 解决方案 #

若SQL处理灵活性不足，可通过Python读取数据后解析：

import pandas as pd
import pyodbc
import json
from itertools import chain

# 连接SQL Server
conn = pyodbc.connect(
    'DRIVER={ODBC Driver 17 for SQL Server};'
    'SERVER=你的服务器地址;'
    'DATABASE=目标数据库;'
    'UID=用户名;'
    'PWD=密码'
)

# 读取原表数据
df = pd.read_sql("SELECT _id, observacoes FROM my.table", conn)
conn.close()

# 定义单个JSON数组的解析函数
def parse_json_item(json_str, row_id):
    result_list = []
    try:
        json_arr = json.loads(json_str)
        for item in json_arr:
            # 提取固定字段
            created_at = item.get('created_at')
            from_obj = item.get('from', {})
            from_name = from_obj.get('name')
            from_id = from_obj.get('userId')
            
            # 提取动态文本字段
            text = None
            for key in item:
                if '-parecer' in key:
                    text = item[key].get(f"{key}-texto")
                    break
            
            result_list.append({
                'row_id': row_id,
                'created_at': created_at,
                'from_name': from_name,
                'from_id': from_id,
                'text': text
            })
    except Exception as e:
        print(f"处理row_id {row_id}时出错: {str(e)}")
    return result_list

# 批量处理所有行
all_results = list(chain.from_iterable(df.apply(lambda x: parse_json_item(x['observacoes'], x['_id']), axis=1)))

# 转换为DataFrame
result_df = pd.DataFrame(all_results)
print(result_df.head())

# 可选：保存结果到CSV或数据库
# result_df.to_csv('extracted_data.csv', index=False)

关键逻辑： #

1. 用pyodbc连接SQL Server读取原始数据；
2. 自定义函数遍历每个JSON对象，提取固定字段；
3. 遍历对象键名，匹配-parecer后缀的动态键，提取对应文本；
4. 合并所有结果为DataFrame，方便后续分析或存储。

内容的提问来源于stack exchange，提问作者Vinícius Sodré Quadros