将 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 数据的原因:
- API 集成:大多数 REST API 期望 JSON 负载,而不是 CSV
- JavaScript 兼容性:JSON 是 JavaScript 原生的,使其成为 Web 应用程序的理想选择
- 分层数据:JSON 支持 CSV 无法表示的嵌套结构
- 类型保留:JSON 区分字符串、数字、布尔值和空值
- 数据交换:JSON 在不同编程语言和平台之间更具可移植性
数据类型转换挑战
将 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 或其他字符集进行编码。编码不匹配会导致文本乱码,尤其是对于非英语字符:
- UTF-8:现代标准,支持所有 Unicode 字符
- Latin-1(ISO-8859-1):在较旧的欧洲系统中常见
- Windows-1252:微软对 Latin-1 的扩展
- UTF-16:某些 Excel 导出使用
读取 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 = []