将嵌套 JSON 转换为 CSV:Python 开发者的实战指南

在日常的数据处理工作中,我们经常需要面对各种格式的数据交换。JSON(JavaScript Object Notation)因其轻量级和易于解析的特性,成为了 Web 开发和 API 接口中的事实标准。然而,当我们需要进行数据分析、生成报表或者将数据导入传统的数据库系统时,CSV(Comma-Separated Values)格式往往因其简洁性和与 Excel 等工具的完美兼容性而更受青睐。

在这篇文章中,我们将深入探讨如何在 Python 中将复杂的嵌套 JSON 结构转换为扁平化的 CSV 表格。无论你是正在处理简单的配置文件,还是面对从 API 获取的海量嵌套数据,通过本文的实战案例和代码解析,你都将掌握一套高效、健壮的处理方案。

理解数据结构:从简单到嵌套

在动手写代码之前,让我们先厘清 JSON 数据的结构演变。这有助于我们理解为什么“转换”这一步并不总是像看起来那么简单。

#### 简单的 JSON 结构

最基础的 JSON 文件通常由键值对组成,类似于 Python 中的字典结构。让我们看一个直观的例子:

{
    "name": "Amit Pathak",
    "profile": "Software Engineer",
    "age": 24,
    "location": "London, UK"
}

在这个简单的例子中,数据结构是扁平的。键直接对应值,我们可以很容易地将这些键映射为 CSV 的表头,将值映射为一行数据。这里的“name”、“profile”和“location”都是顶级键,没有任何层级嵌套。

#### 嵌套 JSON 的挑战

现实世界的数据往往更加复杂。为了表达更丰富的层级关系,JSON 允许值的本身也是一个 JSON 对象。这就是所谓的“嵌套 JSON”。

请看下面这个稍微复杂一点的例子:

{
    "article_id": 3214507,
    "article_link": "http://sample.link",
    "source": "moneycontrol",
    "article": {
        "title": "IT stocks to see a jump this month",
        "category": "finance",
        "sentiment": "neutral"
    }
}

在这个结构中,键“article”的值并不是一个简单的字符串或数字,而是另一个包含了“title”、“category”等信息的字典。

问题来了: CSV 是一种二维表格格式,它天然不支持多层级结构。如果我们直接尝试将上述 JSON 转换为 CSV,我们很难决定列名该如何定义。是叫“article”?还是将“article”拆解开?这就是我们在编写转换脚本时需要解决的核心问题——数据扁平化

核心策略:手动扁平化与 Python 原生实现

为了应对嵌套结构,我们需要一种策略将多层的字典“压平”成一层。最常见的方法是使用“点号”或“下划线”将父键和子键连接起来。例如,将 INLINECODE1ceac943 或 INLINECODEd02252fd 作为 CSV 中的列名。

下面,我们将通过一个完整的原生 Python 实现,一步步构建一个健壮的转换工具。这种方法的好处是不依赖 Pandas 等第三方库,适合对环境依赖有严格限制的场景。

#### 示例场景

假设我们有一个名为 article.json 的文件,内容如下:

{
    "article_id": 3214507,
    "article_link": "http://sample.link",
    "published_on": "17-Sep-2020",
    "source": "moneycontrol",
    "article": {
        "title": "IT stocks to see a jump this month",
        "category": "finance",
        "image": "http://sample.img",
        "sentiment": "neutral"
    }
}

我们的目标是将其转换为一行 CSV 数据,其中“article”对象内的属性被展开到顶层。

#### 完整代码实现

让我们将任务拆分为四个主要步骤:读取、规范化、生成 CSV、写入文件。

import json

def read_json(filename: str) -> dict:
    """
    读取 JSON 文件并将其转换为 Python 字典对象。
    """
    try:
        with open(filename, "r", encoding=‘utf-8‘) as f:
            data = json.loads(f.read())
    except FileNotFoundError:
        raise Exception(f"错误:找不到文件 {filename}")
    except json.JSONDecodeError:
        raise Exception(f"错误:文件 {filename} 的 JSON 格式无效")
    except Exception as e:
        raise Exception(f"读取 {filename} 时发生未知错误: {str(e)}")

    return data

def normalize_json(data: dict, sep=‘_‘) -> dict:
    """
    核心算法:将嵌套的字典结构扁平化。
    如果值是字典,则将父键和子键连接起来。
    """
    new_data = dict()
    for key, value in data.items():
        if not isinstance(value, dict):
            # 如果不是字典,直接保留
            new_data[key] = value
        else:
            # 如果是字典,递归或拼接键名
            for k, v in value.items():
                # 使用下划线连接父键和子键,例如 article_title
                new_data[f"{key}{sep}{k}"] = v

    return new_data

def generate_csv_data(data: dict) -> str:
    """
    根据扁平化的字典生成 CSV 格式的字符串。
    包含表头和数据行。
    """
    # 定义 CSV 列的顺序,这里保持字典原有的键顺序(Python 3.7+)
    csv_columns = data.keys()

    # 生成 CSV 的第一行:表头
    csv_data = ",".join(csv_columns) + "
"

    # 生成数据行:将所有值转换为字符串并用逗号连接
    new_row = [str(data[col]) for col in csv_columns]
    csv_data += ",".join(new_row) + "
"

    return csv_data

def write_to_file(data: str, filepath: str) -> bool:
    """
    将生成的 CSV 字符串写入到文件中。
    """
    try:
        with open(filepath, "w+", encoding=‘utf-8‘) as f:
            f.write(data)
    except Exception as e:
        raise Exception(f"保存数据到 {filepath} 时出错: {str(e)}")
    
    return True

def main():
    # 1. 读取文件
    data = read_json(filename="article.json")

    # 2. 规范化数据(扁平化)
    new_data = normalize_json(data=data)
    print("处理后的扁平化字典:", new_data)

    # 3. 生成 CSV 格式字符串
    csv_data = generate_csv_data(data=new_data)

    # 4. 写入磁盘
    write_to_file(data=csv_data, filepath="article_output.csv")
    print("转换成功!已生成 article_output.csv")

if __name__ == ‘__main__‘:
    main()

#### 代码解析与实际效果

运行上述代码后,控制台会输出处理后的扁平化字典:

处理后的扁平化字典: {
    ‘article_id‘: 3214507, 
    ‘article_link‘: ‘http://sample.link‘, 
    ‘published_on‘: ‘17-Sep-2020‘, 
    ‘source‘: ‘moneycontrol‘, 
    ‘article_title‘: ‘IT stocks to see a jump this month‘, 
    ‘article_category‘: ‘finance‘, 
    ‘article_image‘: ‘http://sample.img‘, 
    ‘article_sentiment‘: ‘neutral‘
}

请注意发生了什么变化:原本嵌套在 INLINECODE3164d36d 下的 INLINECODE835507cf 现在变成了 article_title。这种扁平化后的结构非常适合直接写入 CSV 文件。

最终生成的 article_output.csv 文件内容如下:

article_id,article_link,published_on,source,article_title,article_category,article_image,article_sentiment
3214507,http://sample.link,17-Sep-2020,moneycontrol,IT stocks to see a jump this month,finance,http://sample.img,neutral

进阶实战:处理更深层级的嵌套

上面的 normalize_json 函数虽然不错,但它有一个局限性:只能处理一层嵌套。如果 JSON 数据中有三层、四层甚至更深层的结构,它就无能为力了。作为专业的开发者,我们需要准备一个更通用的递归方案。

让我们升级一下 normalize_json 函数,使其能够处理任意层级的嵌套。

#### 通用递归扁平化方案

import json

def normalize_json_recursive(data: dict, parent_key=‘‘, sep=‘_‘) -> dict:
    """
    递归地将多层嵌套的 JSON 字典扁平化。
    
    参数:
        data: 输入的 JSON 字典
        parent_key: 父键名(用于递归拼接)
        sep: 连接符,默认为下划线
    """
    items = {}
    for key, value in data.items():
        # 构造新的键名
        new_key = f"{parent_key}{sep}{key}" if parent_key else key
        
        if isinstance(value, dict):
            # 如果值是字典,递归调用自身
            items.update(normalize_json_recursive(value, new_key, sep=sep))
        else:
            # 如果值不是字典,直接赋值
            items[new_key] = value
    
    return items

# 测试数据:多层嵌套
nested_json = """
{
    "user_id": 101,
    "user_info": {
        "name": "Alice",
        "contact": {
            "email": "[email protected]",
            "city": "New York"
        }
    },
    "status": "active"
}
"""

def main_advanced():
    data = json.loads(nested_json)
    
    # 使用通用递归函数
    flat_data = normalize_json_recursive(data)
    print("递归处理后的结果:", flat_data)
    
    # 生成 CSV
    # 注意:这里为了简单演示,我们假设只有一条记录
    csv_headers = ",".join(flat_data.keys())
    csv_row = ",".join([str(v) for v in flat_data.values()])
    
    csv_content = f"{csv_headers}
{csv_row}
"
    
    with open("advanced_output.csv", "w") as f:
        f.write(csv_content)

if __name__ == "__main__":
    main_advanced()

运行结果:

你会发现,即使 INLINECODE0d4007c1 嵌套在 INLINECODE0e47e94c 里面,代码也能完美处理,生成的列名会是 user_info_contact_email。这正是我们在处理复杂 API 响应时需要的灵活性。

实战应用场景与最佳实践

在实际开发中,我们处理的很少是单个对象,而是包含成百上千条记录的 JSON 数组。让我们看看如何处理数组结构,以及一些专业开发中的注意事项。

#### 场景:处理 JSON 对象数组

通常 API 返回的数据格式如下:

[
    {"id": 1, "name": "Item A", "details": {"price": 10, "stock": 100}},
    {"id": 2, "name": "Item B", "details": {"price": 20, "stock": 50}},
    {"id": 3, "name": "Item C", "details": {"price": 15, "stock": 200}}
]

为了处理这种列表结构,我们需要编写一个循环逻辑。关键点在于:我们必须先收集所有的键,以确保 CSV 的表头包含所有可能的字段。因为有些对象可能缺少某些字段,如果直接逐行转换,会导致 CSV 列错位。

import json

def convert_json_array_to_csv(json_list):
    """
    将 JSON 数组转换为 CSV 字符串。
    自动处理不同的键集合,确保所有列对齐。
    """
    # 第一步:收集所有可能的列名(表头)
    # 我们必须遍历所有对象,扁平化它们,并取键的并集
    all_headers = set()
    normalized_data_list = []
    
    for item in json_list:
        # 复用之前的递归扁平化函数
        flat_item = normalize_json_recursive(item) 
        normalized_data_list.append(flat_item)
        all_headers.update(flat_item.keys())
    
    # 将表头排序,以保证输出的一致性(可选)
    headers = sorted(list(all_headers))
    
    # 第二步:构建 CSV 内容
    csv_rows = []
    
    # 添加表头
    csv_rows.append(",".join(headers))
    
    # 添加数据行
    for item in normalized_data_list:
        # 使用 .get(key, ‘‘) 来处理缺失的字段,默认为空字符串
        row = [str(item.get(h, "")) for h in headers]
        # 处理字段中包含逗号的情况,加上引号
        escaped_row = [‘"‘ + field.replace(‘"‘, ‘""‘) + ‘"‘ if ‘,‘ in field else field for field in row]
        csv_rows.append(",".join(escaped_row))
        
    return "
".join(csv_rows)

# 示例运行
json_data = [
    {"id": 1, "name": "Item A", "details": {"price": 10}},
    {"id": 2, "name": "Item B", "details": {"price": 20, "stock": 50}} # 注意这一条多了 stock 字段
]

csv_output = convert_json_array_to_csv(json_data)
print(csv_output)

这个脚本展示了处理数组时的关键细节:动态表头收集。即使第一条记录没有 stock 字段,只要后续记录有,表头里就会包含它,第一行对应的位置会被填为空值。这保证了 CSV 结构的完整性。

常见陷阱与优化建议

在处理这类任务时,有一些经验之谈可以帮你避免很多坑。

  • 编码问题(Encoding):

在 Python 中处理文件时,始终记得显式指定 INLINECODE7f1afe05。当你处理用户生成的内容或国际数据时,缺少这一步可能导致程序抛出 INLINECODE79db9785。

  • CSV 转义字符:

如果你的 JSON 数据中包含逗号(INLINECODE9655c748)或换行符(INLINECODE0a021df2),直接简单的 join 会导致 CSV 格式错乱。根据 RFC 4180 标准,包含这些特殊字符的字段应该用双引号括起来。上面的示例代码中包含了一个基础的转义处理。

  • 性能考量:

如果你正在处理几百兆甚至上 G 的 JSON 大文件,一次性 INLINECODEa53fb8f5 读入内存可能会导致内存溢出(OOM)。在这种情况下,建议使用 INLINECODE474ab919 库进行流式解析,或者分块读取文件。

  • 日期格式处理:

JSON 中的日期通常是字符串(ISO 8601 格式)。如果直接转为 CSV,Excel 可能无法自动识别。你可能需要在转换逻辑中添加日期格式化的步骤,将其转换为 Excel 更友好的格式。

替代方案:使用 Pandas 库

虽然上面的原生实现非常适合理解原理,但在数据科学领域,Pandas 是事实上的标准工具。它极大地简化了这一过程,只需要一行代码即可实现我们刚才写了几十行代码的功能。

import pandas as pd

# 假设 data 是一个包含嵌套 JSON 的字典列表
data = [
    {"id": 1, "info": {"name": "A", "age": 10}},
    {"id": 2, "info": {"name": "B", "age": 20}}
]

df = pd.json_normalize(data)

# 此时 df 已经是一个扁平化的 DataFrame
# 列名会自动变为 info.name, info.age

# 直接保存为 CSV
df.to_csv("pandas_output.csv", index=False)

Pandas 的 json_normalize 函数非常强大,它不仅能处理嵌套,还能处理嵌套数组等极其复杂的结构。如果你的环境中允许安装第三方库,这通常是最高效的选择。

总结

在这篇文章中,我们经历了从简单结构到复杂嵌套,再到大规模数组处理的完整旅程。我们首先手动实现了 JSON 到 CSV 的转换逻辑,深入理解了“扁平化”这一核心概念;随后我们优化了算法以支持递归深度的嵌套;最后,我们探讨了处理数组时的对齐问题以及 Pandas 这一强大的替代工具。

掌握这些技能后,你可以更加自信地处理各种 messy 的数据源,将非结构化的 JSON 数据转化为易于分析、可视化和存储的 CSV 表格。无论你是要生成报表、迁移数据还是进行机器学习预处理,这些技术都是你工具箱中不可或缺的一部分。希望这篇指南能帮助你在实际项目中游刃有余!

声明:本站所有文章,如无特殊说明或标注,均为本站原创发布。任何个人或组织,在未征得本站同意时,禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益,可联系我们进行处理。如需转载,请注明文章出处豆丁博客和来源网址。https://shluqu.cn/40699.html
点赞
0.00 平均评分 (0% 分数) - 0