数据科学中的 JSON：Python 和 Pandas 指南
JSON 在数据科学工作流中的完整指南。学习用 Python、Pandas 处理 JSON 并集成到 ML 管道中。
Big JSON Team
• Technical WriterExpert in JSON data manipulation, API development, and web technologies. Passionate about creating tools that make developers' lives easier.
# 数据科学中的 JSON：Python 和 Pandas 指南
JSON 在数据科学中无处不在。它用于 API 数据、NoSQL 数据库、配置文件和机器学习管道。本指南展示如何在数据科学工作流中有效地使用 JSON。
JSON 在数据科学中的角色
常见用途
为什么 JSON 很重要
- ✅ 灵活 - 支持嵌套和不同类型
- ✅ 标准 - 广泛支持的格式
- ✅ 可读 - 人类可读的结构
- ✅ 轻量 - 比 XML 更紧凑
使用 Pandas 加载 JSON
基本加载
import pandas as pd
# 从文件加载
df = pd.read_json('data.json')
# 从字符串加载
df = pd.read_json(json_string)
# 从 URL 加载
df = pd.read_json('https://api.example.com/data')
# 查看数据
print(df.head())
print(df.info())
JSON 行格式
JSON Lines（或 JSONL）是每行一个 JSON 对象的文本格式。非常适合流式处理。
# JSON Lines 文件
# {"name": "张三", "age": 30}
# {"name": "李四", "age": 25}
df = pd.read_json('data.jsonl', lines=True)
orient 参数
JSON 可以有不同的结构：
# 记录导向（数组中的对象）
# [{"name": "张三", "age": 30}, ...]
df = pd.read_json(json_str, orient='records')
# 列导向
# {"name": ["张三", "李四"], "age": [30, 25]}
df = pd.read_json(json_str, orient='columns') # 默认
# 索引导向
# {"0": {"name": "张三", "age": 30}, ...}
df = pd.read_json(json_str, orient='index')
# 分割导向
# {"index": [0, 1], "columns": ["name", "age"], "data": [...]}
df = pd.read_json(json_str, orient='split')
处理嵌套 JSON
大多数现实世界的 JSON 是嵌套的。Pandas 提供了处理嵌套结构的工具。
json_normalize 用于简单嵌套
import pandas as pd
# 嵌套数据
data = {
"users": [
{
"name": "张三",
"age": 30,
"address": {
"city": "北京",
"zip": "100000"
}
},
{
"name": "李四",
"age": 25,
"address": {
"city": "上海",
"zip": "200000"
}
}
]
}
# 展平嵌套数据
df = pd.json_normalize(data['users'])
# 结果列：
# name age address.city address.zip
# 张三 30 北京 100000
# 李四 25 上海 200000
处理深度嵌套的数据
对于深度嵌套的结构，使用递归函数：
import pandas as pd
import json
def flatten_json(nested_json, sep='_'):
"""递归展平嵌套 JSON"""
out = {}
def flatten(x, name=''):
if type(x) is dict:
for a in x:
flatten(x[a], name + a + sep)
elif type(x) is list:
i = 0
for item in x:
flatten(item, name + '[' + str(i) + ']' + sep)
i += 1
else:
out[name[:-1]] = x
flatten(nested_json)
return out
# 使用
data = json.load(open('deeply_nested.json'))
flat_data = flatten_json(data)
df = pd.DataFrame([flat_data])
从 API 加载数据
简单 API 请求
import pandas as pd
import requests
# 获取数据
response = requests.get('https://api.example.com/users')
data = response.json()
# 转换为 DataFrame
df = pd.DataFrame(data)
# 如果响应是嵌套的
if isinstance(data, dict):
df = pd.json_normalize(data['results'])
elif isinstance(data, list):
df = pd.DataFrame(data)
处理分页
import pandas as pd
import requests
def load_paginated_api(url, page_param='page'):
"""从分页 API 加载所有数据"""
all_data = []
page = 1
while True:
response = requests.get(url, params={page_param: page})
data = response.json()
if not data or (isinstance(data, dict) and 'results' not in data):
break
# 处理不同的 API 格式
results = data.get('results', data)
all_data.extend(results if isinstance(results, list) else [results])
page += 1
return pd.DataFrame(all_data)
# 使用
df = load_paginated_api('https://api.example.com/users')
API 认证
import pandas as pd
import requests
# 使用 API 密钥
headers = {
'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'
}
response = requests.get('https://api.example.com/data', headers=headers)
df = pd.DataFrame(response.json())
处理缺失数据
JSON 可能有不一致的结构或缺失字段。
缺失字段
import pandas as pd
# 数据中缺失某些字段
data = [
{"name": "张三", "age": 30, "email": "zhang@example.com"},
{"name": "李四", "age": 25}, # 缺少 email
{"name": "王五", "email": "wang@example.com"} # 缺少 age
]
df = pd.DataFrame(data)
# 填充缺失值
df['email'] = df['email'].fillna('unknown')
df['age'] = df['age'].fillna(df['age'].mean())
# 或删除缺失值
df = df.dropna()
不一致的嵌套
import pandas as pd
# 某些记录有嵌套的地址，有些没有
data = [
{"name": "张三", "address": {"city": "北京"}},
{"name": "李四"} # 没有地址
]
# 处理不一致性
def extract_city(record):
if isinstance(record.get('address'), dict):
return record['address'].get('city')
return None
df = pd.DataFrame(data)
df['city'] = df.apply(lambda row: extract_city(row), axis=1)
大文件处理
对于超过内存的大 JSON 文件，使用流式处理。
JSON Lines 分块读取
import pandas as pd
# 分块读取大的 JSON Lines 文件
chunks = pd.read_json('large.jsonl', lines=True, chunksize=10000)
for chunk in chunks:
# 处理每个块
process_data(chunk)
使用 ijson 库
ijson 允许流式解析 JSON，无需将整个文件加载到内存中。
pip install ijsonimport ijson
def process_large_json(filename, item_path='items.item'):
"""流式处理大 JSON 文件"""
with open(filename, 'rb') as f:
for item in ijson.items(f, item_path):
yield item
# 使用
for record in process_large_json('huge.json'):
print(record)
并行处理
import pandas as pd
from multiprocessing import Pool
def process_chunk(chunk):
# 对每个块执行分析
return chunk.groupby('category').size()
# 读取分块
chunks = pd.read_json('data.jsonl', lines=True, chunksize=10000)
# 并行处理
with Pool(4) as pool:
results = pool.map(process_chunk, chunks)
# 合并结果
final_result = pd.concat(results)
数据类型处理
日期解析
import pandas as pd
# 自动解析日期
df = pd.read_json('data.json', parse_dates=['created_at', 'updated_at'])
# 或手动指定
df['created_at'] = pd.to_datetime(df['created_at'])
分类数据
import pandas as pd
df = pd.read_json('data.json')
# 将字符串列转换为分类
df['category'] = df['category'].astype('category')
df['status'] = df['status'].astype('category')
# 内存节省
print(df.memory_usage())
将数据导出为 JSON
导出到文件
import pandas as pd
df = pd.read_csv('data.csv')
# 默认导出
df.to_json('output.json')
# 以记录格式导出
df.to_json('output.json', orient='records')
# 漂亮打印
df.to_json('output.json', indent=2)
# JSON Lines 格式
df.to_json('output.jsonl', orient='records', lines=True)
导出到字符串
# 获取 JSON 字符串
json_str = df.to_json(orient='records')
# 发送到 API
import requests
response = requests.post('https://api.example.com/data', json=json_str)
在机器学习管道中使用 JSON
加载训练数据
import pandas as pd
from sklearn.model_selection import train_test_split
# 从 JSON 加载数据
df = pd.read_json('training_data.json')
# 分离特征和目标
X = df.drop('target', axis=1)
y = df['target']
# 分割数据
X_train, X_test, y_train, y_test = train_test_split(X, y)
保存模型配置
import json
from sklearn.ensemble import RandomForestClassifier
# 模型参数
config = {
'model_type': 'RandomForest',
'n_estimators': 100,
'max_depth': 10,
'random_state': 42
}
# 保存配置
with open('model_config.json', 'w') as f:
json.dump(config, f, indent=2)
# 读取配置
with open('model_config.json', 'r') as f:
config = json.load(f)
# 使用配置创建模型
model = RandomForestClassifier(config)
常见模式和用例
从多个 API 聚合数据
import pandas as pd
import requests
def aggregate_from_apis(api_list):
"""从多个 API 聚合数据"""
all_data = []
for api_url in api_list:
try:
response = requests.get(api_url)
data = response.json()
all_data.extend(data if isinstance(data, list) else [data])
except Exception as e:
print(f"Error fetching {api_url}: {e}")
return pd.DataFrame(all_data)
# 使用
apis = [
'https://api.example.com/users',
'https://api.other.com/customers'
]
df = aggregate_from_apis(apis)
数据质量检查
import pandas as pd
def check_data_quality(df):
"""检查 DataFrame 的数据质量"""
report = {
'total_rows': len(df),
'total_cols': len(df.columns),
'memory_mb': df.memory_usage(deep=True).sum() / 10242,
'missing_values': df.isnull().sum().to_dict(),
'dtypes': df.dtypes.to_dict(),
'duplicates': df.duplicated().sum()
}
return report
# 使用
df = pd.read_json('data.json')
quality = check_data_quality(df)
print(quality)
最佳实践
1. 验证数据加载
# 总是检查数据
df = pd.read_json('data.json')
print(f"形状: {df.shape}")
print(f"列: {df.columns.tolist()}")
print(f"数据类型:
{df.dtypes}")
print(df.head())
2. 处理编码问题
# 指定编码
df = pd.read_json('data.json', encoding='utf-8')
3. 缓存数据
对于频繁使用的数据，缓存加载：
import pickle
# 保存处理后的 DataFrame
df.to_pickle('processed_data.pkl')
# 快速加载
df = pd.read_pickle('processed_data.pkl')
4. 文档化 Schema
为您的 JSON 数据记录预期的 Schema：
# schema.json
{
"name": "string",
"age": "integer",
"email": "string",
"created_at": "datetime",
"tags": "array"
}
故障排除
问题：无法解析日期
# 指定日期列
df = pd.read_json('data.json', parse_dates=['date_column'])
# 或手动转换
df['date'] = pd.to_datetime(df['date'])
问题：列数据类型不正确
# 指定 dtype
df = pd.read_json('data.json', dtype={'id': int, 'code': str})
问题：内存不足
# 使用分块处理
for chunk in pd.read_json('huge.json', lines=True, chunksize=5000):
process_chunk(chunk)
结论
JSON 在数据科学中是不可或缺的。关键技能：
- ✅ 加载 - 使用 Pandas 加载 JSON
- ✅ 处理 - 展平嵌套结构
- ✅ 验证 - 检查数据质量
- ✅ 扩展 - 处理大文件
- ✅ 集成 - 与 ML 管道集成
掌握这些技能，您可以有效地处理任何 JSON 数据！