基于Python递归实现嵌套JSON转多事件字典列表的ETL需求

问题 #

需要将嵌套JSON解析为多个事件字典作为ETL输入，要求用递归处理（避免硬编码字段），为深层嵌套结构添加父级关联ID（ref_id，此处为父级的restaurant_id），输出字典列表而非扁平化DataFrame。

示例输入JSON #

{"event_name": "restaurants","properties": {"_id": "5a9909384309cf90b5739342","name": "Mangal Kebab Turkish Restaurant","restaurant_id": "41009112","borough": "Queens","cuisine": "Turkish","address": {"building": "4620","coord": {"0": -73.9180155,"1": 40.7427742},"street": "Queens Boulevard","zipcode": "11104"},"grades": [{"date": 1414540800000,"grade": "A","score": 12},{"date": 1397692800000,"grade": "A","score": 10},{"date": 1381276800000,"grade": "A","score": 12}]}}

期望输出字典列表 #

[{"event_name": "restaurants","properties": {"restaurant_id": "41009112","name": "Mangal Kebab Turkish Restaurant","cuisine": "Turkish","_id": "5a9909384309cf90b5739342","borough": "Queens"}},{"event_name": "restaurant_address","properties": {"zipcode": "11104","ref_id": "41009112","street": "Queens Boulevard","building": "4620"}},{"event_name": "restaurant_address_coord","ref_id": "41009112","0": -73.9180155,"1": 40.7427742},{"event_name": "restaurant_grades","properties": {"date": 1414540800000,"ref_id": "41009112","score": 12,"grade": "A","index": "0"}},{"event_name": "restaurant_grades","properties": {"date": 1397692800000,"ref_id": "41009112","score": 10,"grade": "A","index": "1"}},{"event_name": "restaurant_grades","properties": {"date": 1381276800000,"ref_id": "41009112","score": 12,"grade": "A","index": "2"}}]

解决方案 #

以下是Python递归实现的解析逻辑，核心思路为递归遍历JSON结构，区分基础字段、嵌套字典、列表类型，传递父级关联ID并生成独立事件：

实现代码 #

def parse_nested_json(data, ref_id=None, parent_event=None, results=None):
    if results is None:
        results = []
    
    # 处理根节点
    if parent_event is None:
        event_name = data['event_name']
        properties = data['properties'].copy()
        current_ref_id = properties.get('restaurant_id')
        # 分离基础字段与嵌套结构
        nested_fields = {}
        flat_properties = {}
        for k, v in properties.items():
            if isinstance(v, (dict, list)):
                nested_fields[k] = v
            else:
                flat_properties[k] = v
        # 添加根事件
        results.append({
            'event_name': event_name,
            'properties': flat_properties
        })
        # 递归处理嵌套字段
        for field_name, field_data in nested_fields.items():
            child_event_name = f"{event_name}_{field_name}"
            parse_nested_json(field_data, ref_id=current_ref_id, parent_event=child_event_name, results=results)
        return results
    
    # 处理嵌套字典
    if isinstance(data, dict):
        event = {'event_name': parent_event}
        if ref_id is not None:
            # 根据子结构是否有嵌套，决定是否用properties包裹
            if all(not isinstance(v, (dict, list)) for v in data.values()):
                event['ref_id'] = ref_id
                event.update(data)
            else:
                properties = data.copy()
                properties['ref_id'] = ref_id
                event['properties'] = properties
        results.append(event)
        # 递归处理字典内的嵌套字段
        for field_name, field_data in data.items():
            if isinstance(field_data, (dict, list)):
                child_event_name = f"{parent_event}_{field_name}"
                parse_nested_json(field_data, ref_id=ref_id, parent_event=child_event_name, results=results)
    
    # 处理列表
    elif isinstance(data, list):
        for idx, item in enumerate(data):
            event = {'event_name': parent_event}
            properties = item.copy()
            properties['ref_id'] = ref_id
            properties['index'] = str(idx)
            event['properties'] = properties
            results.append(event)
            # 递归处理列表项内的嵌套结构
            for field_name, field_data in item.items():
                if isinstance(field_data, (dict, list)):
                    child_event_name = f"{parent_event}_{field_name}"
                    parse_nested_json(field_data, ref_id=ref_id, parent_event=child_event_name, results=results)
    
    return results

# 测试示例
sample_json = {"event_name": "restaurants","properties": {"_id": "5a9909384309cf90b5739342","name": "Mangal Kebab Turkish Restaurant","restaurant_id": "41009112","borough": "Queens","cuisine": "Turkish","address": {"building": "4620","coord": {"0": -73.9180155,"1": 40.7427742},"street": "Queens Boulevard","zipcode": "11104"},"grades": [{"date": 1414540800000,"grade": "A","score": 12},{"date": 1397692800000,"grade": "A","score": 10},{"date": 1381276800000,"grade": "A","score": 12}]}}

output = parse_nested_json(sample_json)
import json
print(json.dumps(output, indent=2))

关键逻辑说明 #

1. 根节点处理：提取根事件名，分离基础字段和嵌套字段，生成根事件后递归处理嵌套内容。
2. 嵌套字典处理：生成子事件名，添加ref_id，根据子结构是否有嵌套决定是否用properties包裹。
3. 列表处理：为每个列表项生成独立事件，添加index标识位置，同时保留ref_id关联父级。
4. 递归传递：将父级的ref_id和事件名传递给子结构，保证层级关联。

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