将 CSV 转换为 JSON:方法与陷阱

· 12分钟阅读

目录

理解 CSV 到 JSON 转换的基础知识

将 CSV(逗号分隔值)转换为 JSON(JavaScript 对象表示法)是开发人员遇到的最常见的数据转换任务之一。虽然对于简单数据集来说这个过程看起来很简单,但理解基本机制可以确保您避免可能损坏数据的细微错误。

CSV 文件遵循表格结构,其中第一行通常包含列标题。每个后续行代表一条记录,其值对应于这些标题。相比之下,JSON 使用更灵活和更具表现力的分层键值结构。

基本转换将 CSV 标题映射到 JSON 键,每个数据行成为 JSON 数组中的一个对象:

CSV:
name,age,city
Alice,30,NYC
Bob,25,LA

JSON:
[
  {"name":"Alice","age":"30","city":"NYC"},
  {"name":"Bob","age":"25","city":"LA"}
]

这种一对一的对应关系对于扁平数据结构非常有效。然而,现实场景引入了复杂性:缺失值、不一致的数据类型、特殊字符以及对嵌套结构的需求都需要仔细处理。

专业提示:在转换之前,始终检查 CSV 文件的前几行。查找不一致的分隔符、带引号的字段和可能导致解析错误的意外换行符。

为什么要将 CSV 转换为 JSON?

JSON 已成为 Web API 和现代应用程序开发的事实标准。以下是开发人员经常需要转换 CSV 数据的原因:

数据类型转换挑战

将 CSV 转换为 JSON 时最重要的挑战之一是保留数据类型。CSV 本质上是一种文本格式——每个值都存储为字符串。当您的数据包含需要在 JSON 中正确表示的数字、日期、布尔值或空值时,这会产生问题。

解析数字数据

考虑一个包含产品库存数据的 CSV 文件。如果没有适当的类型转换,价格和数量等数字值仍然是字符串,这会破坏应用程序中的计算和比较。

import csv
import json

def parse_csv_with_types(filename):
    def try_numeric(val):
        # 处理空值
        if not val or val.strip() == '':
            return None
        
        # 首先尝试整数转换
        try:
            return int(val)
        except ValueError:
            pass
        
        # 尝试浮点数转换
        try:
            return float(val)
        except ValueError:
            return val
    
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        data = []
        for row in reader:
            typed_row = {k: try_numeric(v) for k, v in row.items()}
            data.append(typed_row)
    
    return json.dumps(data, indent=2)

这种方法首先尝试将每个值转换为整数,然后转换为浮点数,如果两次转换都失败,最后将其保留为字符串。结果是正确类型的 JSON,可以保留数字精度。

日期和时间处理

日期解析带来了独特的挑战,因为 CSV 文件可以包含无数种格式的日期:ISO 8601、美国格式(MM/DD/YYYY)、欧洲格式(DD/MM/YYYY)或自定义格式。您的转换逻辑需要处理这些变化:

from datetime import datetime

def parse_date(val):
    date_formats = [
        '%Y-%m-%d',           # ISO 格式
        '%m/%d/%Y',           # 美国格式
        '%d/%m/%Y',           # 欧洲格式
        '%Y-%m-%d %H:%M:%S',  # 带时间的 ISO
        '%m/%d/%Y %I:%M %p'   # 带 12 小时制时间的美国格式
    ]
    
    for fmt in date_formats:
        try:
            return datetime.strptime(val, fmt).isoformat()
        except ValueError:
            continue
    
    return val  # 如果没有格式匹配,则返回原始值

快速提示:在处理来自多个来源的日期时,在 JSON 输出中标准化为 ISO 8601 格式(YYYY-MM-DD)。这种格式是明确的,并且作为字符串可以正确排序。

布尔值和空值转换

CSV 文件通常将布尔值表示为"true"/"false"、"yes"/"no"、"1"/"0"或类似的变体。空单元格可能表示空值,但它们也可能是空字符串。您的转换逻辑必须处理这些歧义:

CSV 值 预期类型 常见错误 正确的 JSON
true 布尔值 "true"(字符串) true
1 布尔值或整数 "1"(字符串) true 或 1
(空) 空值 ""(空字符串) null
N/A 空值 "N/A"(字符串) null

使用我们的 CSV 解析器和查看器在转换之前预览您的数据将如何被解释。

处理特殊字符和编码

特殊字符和编码问题导致的转换失败比任何其他问题都多。CSV 文件可能包含字段内的逗号、文本中的换行符、引号或破坏简单解析逻辑的非 ASCII 字符。

带引号的字段和转义字符

CSV 标准(RFC 4180)规定包含逗号、引号或换行符的字段必须用双引号括起来。在带引号的字段中,引号本身必须通过加倍来转义:

name,description,price
"Widget A","A simple, reliable widget",19.99
"Widget ""Pro""","The ""best"" widget available",49.99

强大的 CSV 解析器会自动处理这些情况。如果您正在编写自己的解析器,则需要跟踪您是否在带引号的字段内,并正确处理转义序列。

字符编码问题

CSV 文件可以使用 UTF-8、Latin-1、Windows-1252 或其他字符集进行编码。编码不匹配会导致文本乱码,尤其是对于非英语字符:

读取 CSV 文件时始终明确指定编码:

import csv
import json

def convert_with_encoding(filename, encoding='utf-8'):
    try:
        with open(filename, 'r', encoding=encoding) as f:
            reader = csv.DictReader(f)
            data = list(reader)
            return json.dumps(data, ensure_ascii=False, indent=2)
    except UnicodeDecodeError:
        # 尝试替代编码
        for alt_encoding in ['latin-1', 'windows-1252', 'utf-16']:
            try:
                with open(filename, 'r', encoding=alt_encoding) as f:
                    reader = csv.DictReader(f)
                    data = list(reader)
                    return json.dumps(data, ensure_ascii=False, indent=2)
            except UnicodeDecodeError:
                continue
        raise ValueError(f"无法使用任何已知编码解码 {filename}")

专业提示:json.dumps() 中的 ensure_ascii=False 参数保留输出中的 Unicode 字符,而不是将它们转义为 \uXXXX 序列,使 JSON 更具可读性。

字节顺序标记(BOM)

某些应用程序,特别是 Microsoft Excel,会在 UTF-8 文件的开头添加字节顺序标记(BOM)。这个不可见的字符可能导致第一个字段名被误读。Python 的编码参数使用 'utf-8-sig' 自动处理此问题:

with open(filename, 'r', encoding='utf-8-sig') as f:
    reader = csv.DictReader(f)

从扁平 CSV 创建嵌套 JSON 结构

CSV 本质上是扁平的——它表示二维表。JSON 支持具有嵌套对象和数组的分层结构。将扁平 CSV 数据转换为嵌套 JSON 需要深思熟虑的设计和额外的逻辑。

分组相关数据

考虑一个包含客户订单的 CSV 文件,其中每行都为每个订单重复客户信息:

customer_id,customer_name,order_id,product,quantity
101,Alice,1001,Widget,5
101,Alice,1002,Gadget,3
102,Bob,1003,Widget,2

更好的 JSON 结构将订单分组到每个客户下:

[
  {
    "customer_id": 101,
    "customer_name": "Alice",
    "orders": [
      {"order_id": 1001, "product": "Widget", "quantity": 5},
      {"order_id": 1002, "product": "Gadget", "quantity": 3}
    ]
  },
  {
    "customer_id": 102,
    "customer_name": "Bob",
    "orders": [
      {"order_id": 1003, "product": "Widget", "quantity": 2}
    ]
  }
]

以下是如何实现此转换:

import csv
import json
from collections import defaultdict

def csv_to_nested_json(filename):
    customers = defaultdict(lambda: {"orders": []})
    
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            customer_id = int(row['customer_id'])
            
            # 如果尚未设置,则设置客户信息
            if 'customer_id' not in customers[customer_id]:
                customers[customer_id]['customer_id'] = customer_id
                customers[customer_id]['customer_name'] = row['customer_name']
            
            # 添加订单
            customers[customer_id]['orders'].append({
                'order_id': int(row['order_id']),
                'product': row['product'],
                'quantity': int(row['quantity'])
            })
    
    return json.dumps(list(customers.values()), indent=2)

嵌套键的点表示法

另一种方法在 CSV 标题中使用点表示法来指示嵌套:

name,address.street,address.city,address.zip
Alice,123 Main St,NYC,10001
Bob,456 Oak Ave,LA,90001

这将转换为:

[
  {
    "name": "Alice",
    "address": {
      "street": "123 Main St",
      "city": "NYC",
      "zip": "10001"
    }
  }
]

实现需要解析标题键并构建嵌套字典:

def set_nested_value(obj, path, value):
    keys = path.split('.')
    for key in keys[:-1]:
        obj = obj.setdefault(key, {})
    obj[keys[-1]] = value

def csv_to_nested_with_dots(filename):
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        data = []
        for row in reader:
            obj = {}
            for key, value in row.items():
                set_nested_value(obj, key, value)
            data.append(obj)
    return json.dumps(data, indent=2)

大文件扩展和性能优化

转换小型 CSV 文件很简单,但生产系统通常处理包含数百万行的文件。将整个多 GB 的 CSV 加载到内存中会导致崩溃和性能问题。

流式处理

不要将整个文件加载到内存中,而是逐行处理并增量写入 JSON:

import csv
import json

def stream_csv_to_json(input_file, output_file):
    with open(input_file, 'r') as infile, open(output_file, 'w') as outfile:
        reader = csv.DictReader(infile)
        
        outfile.write('[\n')
        first = True
        
        for row in reader:
            if not first:
                outfile.write(',\n')
            first = False
            
            json.dump(row, outfile)
        
        outfile.write('\n]')

这种方法无论文件大小如何都保持恒定的内存使用。权衡是您无法轻松创建嵌套结构或执行需要查看所有数据的聚合。

分块处理

对于需要一些聚合但不需要整个数据集的操作,分块处理文件:

def process_in_chunks(filename, chunk_size=10000):
    with open(filename, 'r') as f:
        reader = csv.DictReader(f)
        chunk = []
        
    

We use cookies for analytics. By continuing, you agree to our Privacy Policy.