嵌套JSON数组标准化失败，求指定格式转换及去重入库方案

使用以下Python代码处理嵌套JSON时，数组无法正常标准化：

import pandas as pd
import json

def on_input(data):
    df = pd.read_json(data)df = pd.json_normalize(json.loads(data))
    api.send("output", df.to_json(orient="records"))

api.set_port_callback("input1", on_input)

尝试过max_level和record_path参数仍无效果。需求如下：

• 将给定的嵌套JSON标准化为指定表格格式
• 支持条件判断（过滤createSource为null的项）
• 字段按顺序添加数字后缀
• 最终数据无重复存入数据库，header.poNumber和data.id为唯一标识

输入JSON

[{
      "header.poNumber":"9023496",
      "data.id":"10013459",
      "message.source":[{
            "createSource":null,
            "timeStamp":"2023-05-12T19:30:00.0000000+02:00",
            "type":"full"
         },{
            "createSource":"testdev",
            "timeStamp":"2023-05-11T19:30:00.0000000+02:00",
            "type":"ordersEstimated"
         },{
            "createSource": "event",
            "timeStamp":"2023-05-12T12:30:00.0000000+01:00",
            "type":"ordersCreated"
         }],
      "message.time":[{
            "timeSource":"UTC",
            "typeId":"full"
         },{
            "timeSource":"IST",
            "typeId":"actual"
         }]
   }]

期望输出

原代码的问题在于直接用json_normalize无法实现自定义的数组过滤和字段重命名逻辑，需要手动处理嵌套数组：

import pandas as pd
import json

def on_input(data):
    # 加载JSON数据
    json_data = json.loads(data)
    processed_items = []
    
    for item in json_data:
        # 提取主字段
        base_info = {
            "header.poNumber": item["header.poNumber"],
            "data.id": item["data.id"]
        }
        
        # 过滤message.source中createSource不为null的项
        filtered_sources = [src for src in item["message.source"] if src["createSource"] is not None]
        # 按顺序添加带数字后缀的字段
        for idx, src in enumerate(filtered_sources, start=1):
            base_info[f"message.source.timestamp{idx}"] = src["timeStamp"]
            base_info[f"message.source.type{idx}"] = src["type"]
        
        processed_items.append(base_info)
    
    # 转换为DataFrame
    df = pd.DataFrame(processed_items)
    # 去重，基于唯一标识字段
    df = df.drop_duplicates(subset=["header.poNumber", "data.id"], keep="last")
    
    # 输出处理后的JSON
    api.send("output", df.to_json(orient="records"))

api.set_port_callback("input1", on_input)

代码说明 #

1. 加载与遍历数据：将输入的JSON字符串转为Python对象，逐个处理每个顶层对象
2. 提取主字段：先提取header.poNumber和data.id这些核心标识字段
3. 过滤与处理嵌套数组：过滤掉createSource为null的无效项，给有效项的字段添加数字后缀后合并到主信息中
4. 去重处理：基于唯一标识字段执行去重，保留最新记录
5. 输出结果：将处理后的DataFrame转为指定格式的JSON输出

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