如何写数据格式转换脚本

wen 实用脚本 2026-06-30 27

从入门到自动化实战

📖 目录导读

什么是数据格式转换脚本？为什么要学它？
常见数据格式及其应用场景
写转换脚本前的准备工作
手把手写一个JSON转CSV的Python脚本
进阶：处理大文件与多格式互转技巧
常见错误排查与性能优化
Q&A 环节

什么是数据格式转换脚本？为什么要学它？

数据格式转换脚本，本质上是一段程序，能将一种数据结构（如JSON、XML、CSV、YAML、Excel等）解析后，按目标格式的规范重新组织并输出，举个直观的例子：你从API拿到了一堆JSON格式的用户信息，但老板要求用Excel表格统计——手动复制几千条数据显然不现实,这时一个转换脚本就能在几秒钟内完成任务。

如何写数据格式转换脚本

为什么你需要掌握这项技能？

数据工程师日常：80%工作涉及ETL（抽取-转换-加载），格式转换是核心环节

避免重复劳动：一次编写，永久复用，还可扩展支持新格式

减少人工错误：手动处理CSV时常见的编码问题、字段错位都能通过脚本避免

职业竞争力：在招聘网站搜索“数据格式转换脚本”，大量数据分析、运维岗位明确要求此技能

关键认知：转换脚本的核心不是“翻译”，而是数据映射——即如何将源数据的字段准确对应到目标格式的字段，比如JSON可能有嵌套结构，而CSV只有扁平表格，这时需要设计“展平”策略。

常见数据格式及其应用场景

格式	特点	典型场景
CSV	纯文本、体积小、易打开	数据分析、Excel导入导出
JSON	结构灵活、支持嵌套	API接口、NoSQL数据库
XML	可扩展、有属性	旧系统配置文件、SOAP服务
YAML	可读性强、适合配置	容器编排(Docker)、CI/CD文件
Parquet	列式存储、压缩率高	大数据分析、Hadoop生态

注意：并非所有格式都能无损失互转，例如JSON中的数组无法直接映射到CSV的单个单元格,需要决定是展开为多行还是合并为字符串。

写转换脚本前的准备工作

在动手敲代码之前,请先回答以下几个问题：

源数据长什么样？ 拿到一个样例文件，观察其结构、字段名、数据类型
目标格式要求什么？ 是严格的标准格式（如RFC 4180定义的CSV）还是特定系统要求的非标准格式？
数据量有多大？ 100行和1000万行的处理策略完全不同
是否需要保留所有字段？ 某些字段可能不需要被转换，或者需要新增计算字段

推荐工具链：

Python 3.x（最通用的选择，拥有对CSV、JSON、XML等格式的原生库支持）
Pandas（处理表格类数据转换的首选库）
jq（针对JSON的命令行神器,适合快速处理）
Apache NiFi（图形化ETL工具,适合企业级复杂转换）

手把手写一个JSON转CSV的Python脚本

假设我们有一个用户数据的JSON文件 users.json：

[
  {"name": "Alice", "age": 30, "address": {"city": "Shanghai", "zip": 200000}},
  {"name": "Bob", "age": 25, "address": {"city": "Beijing", "zip": 100000}}
]

目标：转换为CSV，将嵌套的address字段展平为address_city和address_zip两列。

完整脚本

import json
import csv
def flatten_json(obj, prefix=''):
    """递归展平JSON嵌套，返回扁平的字典"""
    flat_dict = {}
    for key, value in obj.items():
        new_key = f"{prefix}_{key}" if prefix else key
        if isinstance(value, dict):
            flat_dict.update(flatten_json(value, new_key))
        else:
            flat_dict[new_key] = value
    return flat_dict
def json_to_csv(json_file_path, csv_file_path):
    with open(json_file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    if not data:
        print("JSON文件为空")
        return
    # 展平每条记录
    flat_records = [flatten_json(record) for record in data]
    # 提取所有字段名（保持顺序）
    fieldnames = list(flat_records[0].keys())
    with open(csv_file_path, 'w', newline='', encoding='utf-8-sig') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(flat_records)
    print(f"成功转换 {len(flat_records)} 条记录，输出至 {csv_file_path}")
if __name__ == '__main__':
    json_to_csv('users.json', 'users.csv')

运行后生成的CSV：

name,age,address_city,address_zip
Alice,30,Shanghai,200000
Bob,25,Beijing,100000

关键点解析：

使用递归展平嵌套字典，避免硬编码字段深度
utf-8-sig编码确保Excel能正确识别UTF-8文件
通过csv.DictWriter简化字段映射，自动处理引号转义

进阶：处理大文件与多格式互转技巧

大型JSON文件处理（无法一次性加载到内存）

import ijson  # 流式解析JSON，避免内存溢出
def stream_json_to_csv(large_json_file, csv_output, batch_size=1000):
    csv_file = open(csv_output, 'a', newline='', encoding='utf-8')
    writer = None
    with open(large_json_file, 'rb') as f:
        parser = ijson.parse(f)
        for prefix, event, value in parser:
            if event == 'start_map' or (event == 'map_key' and prefix == ''):
                # 此处需根据实际JSON结构编写处理逻辑
                pass

格式互转对照表

源格式	目标格式	推荐方法
CSV → JSON	Pandas + to_json()	处理日期字段时需指定格式
XML → JSON	xmltodict库	注意XML属性与子元素的区别
Excel → CSV	openpyxl或pandas的read_excel	注意合并单元格的处理
YAML → JSON	PyYAML库的dump(json.dumps)	YAML中的None转为JSON的null

实战技巧：使用Pandas一键转换

import pandas as pd
# 几乎支持所有常见格式的互转
df = pd.read_json('input.json')
df.to_csv('output.csv', index=False)      # JSON→CSV
df.to_excel('output.xlsx', index=False)   # JSON→Excel
df.to_parquet('output.parquet')           # JSON→Parquet
# 处理复杂嵌套：先使用json_normalize展平
from pandas import json_normalize
nested_df = json_normalize(data, sep='_')

常见错误排查与性能优化

𝘙常见错误及解决方案

编码乱码：始终声明encoding='utf-8'或'utf-8-sig'，避免使用系统默认编码
字段数不匹配：某些行缺少字段时，使用fieldnames显式声明，并用defaultdict或dict.get()填充默认值
内存溢出：单文件超过内存50%时，改用流式处理或分块读取
特殊字符导致CSV解析错误：确保CSV字段内容包含逗号、换行符时被引号包裹（Python的csv模块会自动处理）

性能优化建议

使用列表推导代替循环：[flatten_json(item) for item in data]比for循环快2-3倍
批量写入：每处理N条记录写入一次文件，而不是每处理一条就flush磁盘
选用合适的数据结构：如果需要频繁查找字段名，用set比list快很多
对超大文件使用多进程：用multiprocessing.Pool分配不同数据块给不同核心

Q&A 环节

Q: 如何转换一个包含中文的JSON到CSV，且Excel打开不乱码？

A: 写入CSV时使用encoding='utf-8-sig'（带有BOM的UTF-8），Excel会自动识别，也可以用df.to_excel()直接输出.xlsx文件。

Q: 我的JSON有数组嵌套，怎么映射到CSV？

A: 三种常见策略：① 将数组元素作为字符串存入一个单元格（如"[1,2,3]"） ② 为每个数组元素创建新行，其他字段重复（适用于一对多关系） ③ 为数组元素创建独立的新列（适用于固定长度数组）,推荐根据业务选择合适的策略。

Q: 有没有不需要写代码的工具？

A: 有，比如免费的DataConverter（[example.com]() 注意：此处原域名已按照要求修改）、Talend Open Studio、阿里DataWorks的数据集成模块，但编写脚本更灵活,且能集成到自动化流水线中。

Q: 脚本运行很慢，怎么定位瓶颈？

A: 用Python的cProfile模块分析：python -m cProfile script.py，关注耗时最长的函数,常见瓶颈是文件IO和重复的类型转换。