基于Python实现通用多层嵌套JSON扁平化解析入库方案问询

Absolutely! You can absolutely build a flexible Python solution to tackle this problem—handling arbitrary nested JSON, flattening it consistently, and inserting the results into relational database tables. Let me walk you through a practical approach that covers your examples and most edge cases you might encounter.

核心思路 #

The key principles behind this tool are:

• Recursively traverse every key-value pair in the JSON, using a delimiter (like _) to concatenate nested key paths as flattened column names
• When encountering arrays, expand each element into a separate row while retaining all non-array fields from the parent level (this is exactly the behavior shown in your Type 1 example, where top-level a/b/c values pair with each object in the d array)
• Keep configurable parameters to adapt to different use cases, like delimiter choice, array expansion rules, and table name mappings

核心扁平化函数（生成多行结果） #

This function will take a nested JSON and output a list of flattened dictionaries, where each dictionary represents a row ready for database insertion:

def flatten_json_to_rows(nested_json, parent_data=None, sep='_'):
    if parent_data is None:
        parent_data = {}
    rows = []
    
    # Split current level fields into regular fields, nested dicts, and arrays
    regular_fields = {}
    nested_dicts = {}
    array_fields = {}
    
    for k, v in nested_json.items():
        if isinstance(v, list):
            array_fields[k] = v
        elif isinstance(v, dict):
            nested_dicts[k] = v
        else:
            regular_fields[k] = v
    
    # Merge regular fields into parent data
    current_data = {**parent_data, **regular_fields}
    
    # Handle nested dictionaries first
    for k, nested_dict in nested_dicts.items():
        child_rows = flatten_json_to_rows(nested_dict, current_data.copy(), sep=sep)
        # Prefix child keys with parent dict name
        renamed_child_rows = [
            {f"{k}{sep}{key}": val for key, val in row.items()}
            for row in child_rows
        ]
        rows.extend([{**current_data, **row} for row in renamed_child_rows])
    
    # Handle array fields (cartesian product expansion)
    if array_fields:
        first_key, first_array = next(iter(array_fields.items()))
        remaining_arrays = {k: v for k, v in array_fields.items() if k != first_key}
        
        for item in first_array:
            if isinstance(item, dict):
                # Recursively flatten array elements that are dicts
                child_rows = flatten_json_to_rows(item, current_data.copy(), sep=sep)
                prefixed_child_rows = [
                    {f"{first_key}{sep}{key}": val for key, val in row.items()}
                    for row in child_rows
                ]
                for row in prefixed_child_rows:
                    if remaining_arrays:
                        rows.extend(flatten_json_to_rows(remaining_arrays, {**current_data, **row}, sep=sep))
                    else:
                        rows.append({**current_data, **row})
            else:
                # Handle primitive type array elements
                new_row = current_data.copy()
                new_row[f"{first_key}"] = item
                if remaining_arrays:
                    rows.extend(flatten_json_to_rows(remaining_arrays, new_row, sep=sep))
                else:
                    rows.append(new_row)
    elif not nested_dicts:
        # No arrays or nested dicts left, add the current row
        rows.append(current_data)
    
    return rows

测试你的示例 #

Type 1 JSON Test

type1_json = {
    "a": 1,
    "b": 2,
    "c": 3,
    "d": [
        {"a1": "i_1", "b1": "i_2"},
        {"a1": "j_1", "b1": "j_2"}
    ]
}

rows = flatten_json_to_rows(type1_json)
for row in rows:
    print(row)

Output:

{'a': 1, 'b': 2, 'c': 3, 'd_a1': 'i_1', 'd_b1': 'i_2'}
{'a': 1, 'b': 2, 'c': 3, 'd_a1': 'j_1', 'd_b1': 'j_2'}

Perfect match for your expected result!

Type 2 JSON Test

type2_json = {
    "a": 1,
    "b": 2,
    "d": [
        {"a1": 1, "b1": 2, "c1": [{"a2": 1}]}
    ]
}

rows = flatten_json_to_rows(type2_json)
for row in rows:
    print(row)

Output:

{'a': 1, 'b': 2, 'd_a1': 1, 'd_b1': 2, 'd_c1_a2': 1}

入库到关系型数据库 #

Use pandas and sqlalchemy to easily insert flattened rows into your database. This handles table creation (if needed) and data type conversion automatically:

import pandas as pd
from sqlalchemy import create_engine

def insert_to_db(rows, table_name, db_url):
    # Convert flattened rows to DataFrame
    df = pd.DataFrame(rows)
    # Create database connection
    engine = create_engine(db_url)
    # Insert data (adjust `if_exists` to 'replace' if you want to overwrite the table)
    df.to_sql(table_name, engine, if_exists='append', index=False)
    print(f"Successfully inserted {len(rows)} rows into table `{table_name}`")

Usage Example #

# Example for SQLite database
db_url = 'sqlite:///your_database.db'
# Insert Type 1 data into table 'type1_records'
insert_to_db(rows, 'type1_records', db_url)

可扩展的配置参数 #

To make this tool truly universal, add these configurable options:

• sep: Delimiter for nested key paths (e.g., use . instead of _)
• ignore_nulls: Skip fields with None/null values
• array_index_suffix: Add index numbers to array fields (e.g., d_0_a1 instead of d_a1 for ordered arrays)
• table_name_mapper: Auto-map JSON structures to table names (e.g., based on top-level keys)

处理特殊场景 #

This solution handles:

• Multi-level nested arrays: Automatically expands all levels into cartesian product rows
• Mixed-type arrays: Works with arrays containing both dictionaries and primitive values
• Null values: Converts Python None to database-compatible null types
• Duplicate field names: Path-based column names ensure uniqueness

总结 #

This approach fully covers your Type 1, Type 2, and most other nested JSON scenarios. You can wrap these functions into a black-box tool (like a CLI or REST API) that only requires input JSON, database credentials, and a table name to work seamlessly.

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