Python DataFrame动态提取嵌套JSON列表字段技术问询

有一个包含Identifier和Properties两列的Python DataFrame，其中Properties列存储的是嵌套JSON列表，且不同行的数据结构异构（比如前5行只有sms和file类型数据，第6行包含account、file、process、host等多种类型）。需要实现动态提取指定字段（如sms的SMName、file的Name/hash、account的ETNDomain/UserPrincipalName等）生成独立列，适配异构数据的差异。

步骤1：解析JSON并处理$ref引用 #

原始JSON中包含$ref引用，需要先递归解析这些引用，把嵌套结构扁平化，避免数据丢失：

import json
import pandas as pd

def resolve_refs(json_obj):
    # 先收集所有带$id的对象，建立ID映射
    id_map = {}
    def collect_ids(item):
        if isinstance(item, dict):
            if "$id" in item:
                id_map[item["$id"]] = item
            for k, v in item.items():
                collect_ids(v)
        elif isinstance(item, list):
            for elem in item:
                collect_ids(elem)
    collect_ids(json_obj)
    
    # 替换$ref为实际对象
    def replace_refs(item):
        if isinstance(item, dict):
            if "$ref" in item:
                ref_id = item["$ref"]
                return id_map.get(ref_id, item)
            new_item = {}
            for k, v in item.items():
                new_item[k] = replace_refs(v)
            return new_item
        elif isinstance(item, list):
            return [replace_refs(elem) for elem in item]
        else:
            return item
    
    return replace_refs(json_obj)

步骤2：动态提取异构字段 #

针对每行的JSON列表，按Type分类提取目标字段，生成结构化字典后合并到原DataFrame：

def extract_properties(props_json):
    # 解析并处理引用
    resolved = resolve_refs(props_json)
    result = {}
    for item in resolved:
        item_type = item.get("Type")
        if not item_type:
            continue
        # 按类型提取目标字段，可按需扩展
        if item_type == "sms":
            result["sms_SMName"] = item.get("SMName")
        elif item_type == "file":
            result["file_Name"] = item.get("Name")
            # 提取不同算法的哈希值
            if "hash" in item:
                for h in item["hash"]:
                    result[f"file_hash_{h['Algorithm']}"] = h.get("Value")
            elif "FileHashes" in item:
                for h in item["FileHashes"]:
                    result[f"file_hash_{h['Algorithm']}"] = h.get("Value")
        elif item_type == "account":
            result["account_ETNDomain"] = item.get("ETNDomain")
            result["account_UserPrincipalName"] = item.get("UserPrincipalName")
            result["account_Name"] = item.get("Name")
        elif item_type == "host":
            result["host_FQDN"] = item.get("FQDN")
            result["host_RiskScore"] = item.get("RiskScore")
        # 可添加更多类型的字段提取逻辑，比如process类型的CommandLine等
    return result

# 应用到原始DataFrame
# 先将Properties列的字符串转为JSON对象
df["Properties"] = df["Properties"].apply(json.loads)
# 提取字段生成临时DataFrame
extracted_df = df["Properties"].apply(extract_properties).apply(pd.Series)
# 合并到原始DataFrame
final_df = pd.concat([df[["Identifier"]], extracted_df], axis=1)

关键说明 #

• 自动适配异构数据：某行不存在的字段会填充为NaN，不破坏整体数据结构
• 可按需扩展extract_properties函数，增加更多类型或字段的提取逻辑
• 处理$ref是核心步骤，否则会丢失引用的嵌套数据

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