在电商 API 开发中，数据解析与存储是实现业务逻辑的核心环节。本文将系统拆解从 API 返回的 JSON 数据到数据库持久化的全流程，涵盖数据提取、清洗、转换、存储及优化策略，结合 Python 实战代码与主流数据库方案，帮助开发者构建健壮的数据管道。
一、数据解析流程概述
- 数据解析核心步骤
plaintext
API响应（JSON） → 数据提取 → 结构解析 → 类型转换 → 数据验证 → 数据库存储 - 关键技术栈
解析工具：Python json 模块、ijson（流式解析）、marshmallow（数据验证）
关系型数据库：SQLAlchemy（MySQL/PostgreSQL）、异步驱动 asyncpg
非关系型数据库：MongoDB、异步驱动 motor
数据清洗：pandas、SQL 语句（去重、格式修正）
二、JSON 数据解析实战 - 基础解析：使用 Python 内置库
python
import json
解析JSON响应
response = '{"name": "iPhone 15", "price": 7999.99, "category": ["手机", "数码"]}'
data = json.loads(response)
提取字段
product_name = data["name"]
product_price = data["price"]
product_categories = data["category"]
- 嵌套结构处理：递归解析
python
def parse_nested_data(data, prefix=""):
parsed = {}
for key, value in data.items():
return parsednew_key = f"{prefix}_{key}" if prefix else key if isinstance(value, dict): parsed.update(parse_nested_data(value, new_key)) elif isinstance(value, list) and all(isinstance(item, dict) for item in value): for idx, item in enumerate(value): parsed.update(parse_nested_data(item, f"{new_key}_{idx}")) else: parsed[new_key] = value
示例嵌套数据
nested_data = {
"product": {
"id": 1001,
"specs": {
"color": "black",
"storage": "256GB"
},
"reviews": [{"rating": 4.8}, {"rating": 4.5}]
}
}
parsed_data = parse_nested_data(nested_data)
- 大文件处理：流式解析
python
import ijson
处理10GB级JSON文件
with open('large_data.json', 'r') as f:
parser = ijson.parse(f)
for prefix, event, value in parser:
if prefix.endswith('price') and event == 'number':
print(f"价格: {value}")
三、数据清洗与标准化
- 缺失值处理
python方法1：填充默认值
cleaned_data = {k: v if v is not None else "N/A" for k, v in raw_data.items()}
方法2：SQL语句更新（MySQL）
update_sql = """
UPDATE products
SET price = COALESCE(price, 0)
WHERE price IS NULL
"""
- 重复数据删除
python关系型数据库：使用窗口函数
delete_duplicates_sql = """
WITH CTE AS (
SELECT *,
FROM productsROW_NUMBER() OVER (PARTITION BY sku_id ORDER BY update_time DESC) AS row_num
)
DELETE FROM CTE WHERE row_num > 1
"""
MongoDB：使用聚合框架
db.products.aggregate([
{"$group": {"_id": "$sku_id", "unique_ids": {"$addToSet": "$_id"}}},
{"$match": {"unique_ids": {"$size": {"$gt": 1}}}},
{"$out": "duplicates"}
])
- 格式标准化
python统一日期格式
from datetime import datetime
def parse_date(date_str):
try:
return datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%S").strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
return None
修正价格格式
cleaned_price = float(str(raw_price).replace("¥", "").replace(",", ""))
四、数据存储策略
- 关系型数据库存储：SQLAlchemy 示例
python
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
定义数据模型
Base = declarative_base()
class Product(Base):
tablename = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(255))
price = Column(Float)
category = Column(String(255))
批量插入
engine = create_engine('mysql+pymysql://user:password@localhost/ecommerce')
Session = sessionmaker(bind=engine)
session = Session()
products = [Product(name=p["name"], price=p["price"], category=p["category"]) for p in parsed_data]
session.bulk_save_objects(products)
session.commit()
- 非关系型数据库存储：MongoDB 示例
python
from pymongo import MongoClient
连接MongoDB
client = MongoClient('mongodb://localhost:27017')
db = client['ecommerce']
collection = db['products']
处理嵌套数据
product_data = {
"name": "iPhone 15",
"price": 7999.99,
"specs": {
"color": "black",
"storage": "256GB"
},
"categories": ["手机", "数码"]
}
collection.insert_one(product_data)
- 异步存储：结合 aiohttp 与 motor
python
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient
async def async_store_data(data):
client = AsyncIOMotorClient('mongodb://localhost:27017')
db = client['ecommerce']
await db.products.insert_one(data)
await client.close()
配合异步请求
async def fetch_and_store(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.json()
await async_store_data(data)
五、数据验证与异常处理
- 使用 Marshmallow 定义数据模式
python
from marshmallow import Schema, fields, validate
class ProductSchema(Schema):
name = fields.Str(required=True, validate=validate.Length(min=2))
price = fields.Float(required=True, validate=validate.Range(min=0))
category = fields.List(fields.Str(), required=True)
created_at = fields.DateTime(required=True)
验证数据
schema = ProductSchema()
validated_data = schema.load(raw_data)
- 异常处理机制
python
try:
parsed_data = json.loads(response)
validated_data = ProductSchema().load(parsed_data)
await async_store_data(validated_data)
except json.JSONDecodeError as e:
logging.error(f"JSON解析失败: {str(e)}")
except ValidationError as e:
logging.error(f"数据验证失败: {e.messages}")
except Exception as e:
logging.error(f"存储失败: {str(e)}")
六、性能优化策略 - 批量插入与事务
python关系型数据库：批量插入
session.bulk_save_objects(products) # 比逐条插入快10倍以上
MongoDB：批量写入
collection.insert_many(products_list)
- 索引优化
python关系型数据库：添加索引
CREATE INDEX idx_product_price ON products (price);
MongoDB：创建复合索引
db.products.create_index([("category", 1), ("price", -1)])
- 异步处理与连接池
python使用asyncpg异步写入PostgreSQL
import asyncpg
async def async_store_postgres(data):
conn = await asyncpg.connect(user='user', password='password', database='ecommerce')
await conn.execute(
"INSERT INTO products (name, price, category) VALUES ($1, $2, $3)",
data["name"], data["price"], data["category"]
)
await conn.close()
- 大文件处理：流式解析与分批存储
python
import ijson
async def stream_and_store(file_path):
with open(file_path, 'r') as f:
parser = ijson.items(f, 'item')
batch = []
for item in parser:
batch.append(item)
if len(batch) >= 1000:
await async_store_data(batch)
batch = []
if batch:
await async_store_data(batch)
七、数据模型设计最佳实践
- 关系型数据库：扁平化设计
python原始嵌套数据
{
"product": {
}"id": 1001, "specs": { "color": "black", "storage": "256GB" }
}
扁平化后表结构
products: id, name, color, storage
- 非关系型数据库：内嵌与引用
python内嵌模型（适合数据关联性强）
{
"product_id": 1001,
"specs": {
}"color": "black", "storage": "256GB"
}
引用模型（适合数据独立性高）
products: product_id, name
specs: spec_id, color, storage, product_id
八、监控与调优
- 关键指标监控
解析耗时：记录 JSON 解析时间，定位慢解析点
存储吞吐量：使用 Prometheus 监控数据库写入 QPS
错误率：统计解析失败、验证失败、存储失败的比例 - 调优工具推荐
数据管道：Apache Airflow（任务调度）、Luigi（工作流管理）
性能分析：Py-Spy（追踪 Python 性能瓶颈）、MongoDB Atlas（数据库性能监控）
九、总结
通过本文的实战指南，开发者可以掌握以下核心能力：
灵活解析：处理不同复杂度的 JSON 结构，包括嵌套对象与数组
数据清洗：消除缺失值、重复数据、格式不一致等问题
高效存储：根据业务需求选择关系型或非关系型数据库，实现批量插入与异步处理
健壮性保障：使用数据验证库和异常处理机制，确保数据完整性
性能优化：通过索引、连接池、批量操作等技术提升系统吞吐量
通过将这些技术结合到电商 API 开发中，可构建高可用、高性能的数据管道，为价格监控、用户行为分析等业务场景提供坚实基础
