跨境商品数据采集面临格式不统一、字段缺失、编码混乱等问题。不同平台的商品数据，有的用 JSON，有的是 HTML 片段，有的直接是纯文本描述。不做清洗的话，这些数据没法直接入库分析。

本文分享一条完整的 ETL 管线设计，把脏数据变成可用的结构化数据。

## 管线架构

```
原始数据 → 解析层 → 清洗层 → 标准化层 → 校验层 → 存储
```

每层职责单一，通过管道模式串联：

```python
from typing import List, Dict, Any, Callable, Optional
import re, json

class CleanStep:
    """单个清洗步骤"""
    def __init__(self, name: str, fn: Callable):
        self.name = name
        self.fn = fn
    
    def process(self, item: Dict) -> Dict:
        try:
            return self.fn(item)
        except Exception as e:
            item["_error"] = f"{self.name}: {e}"
            return item

class CleaningPipeline:
    """数据清洗管线：可串联多个步骤"""
    
    def __init__(self):
        self.steps: List[CleanStep] = []
    
    def add_step(self, name: str, fn: Callable):
        self.steps.append(CleanStep(name, fn))
    
    def run(self, raw: List[Dict]) -> List[Dict]:
        result = raw
        for step in self.steps:
            print(f"  [清洗] {step.name}...")
            result = [step.process(item) for item in result]
            print(f"    → {len(result)} 条")
        return result
```

## 第一步：字段清洗

从多平台采集的数据，第一个问题是字段空值和不一致：

```python
def clean_fields(item: Dict) -> Dict:
    """去掉空值字段，标准化字段名"""
    cleaned = {}
    field_map = {
        "title": ["title", "name", "product_name", "标题"],
        "price": ["price", "sale_price", "final_price", "价格", "售价"],
        "image": ["image", "img", "picture", "main_image", "主图"],
        "sku": ["sku", "id", "product_id", "item_id", "商品编号"],
    }
    
    for standard_name, aliases in field_map.items():
        for alias in aliases:
            if alias in item and item[alias]:
                cleaned[standard_name] = item[alias]
                break
    
    return cleaned
```

## 第二步：价格标准化

不同平台的价格格式差异很大——"¥99.00"、"$12.99"、"9,800円"、"EUR 15,50"。需要统一转为浮点数：

```python
import re
from decimal import Decimal

CURRENCY_PATTERNS = {
    "CNY": [r"¥\s*([\d,]+\.?\d*)", r"(\d+\.?\d*)\s*元"],
    "USD": [r"\$\s*([\d,]+\.?\d*)", r"USD\s*([\d,]+\.?\d*)"],
    "JPY": [r"([\d,]+)\s*円", r"JPY\s*([\d,]+)"],
    "EUR": [r"([\d,]+\.?\d*)\s*€", r"EUR\s*([\d,]+\.?\d*)"],
}

def normalize_price(raw_price: str) -> Dict:
    """标准化价格为 Decimal，同时保留原始币种"""
    if not raw_price:
        return {"price": 0, "currency": "unknown"}
    
    raw = raw_price.strip()
    for currency, patterns in CURRENCY_PATTERNS.items():
        for pat in patterns:
            m = re.search(pat, raw)
            if m:
                num_str = m.group(1).replace(",", "")
                return {"price": Decimal(num_str), "currency": currency}
    
    # 纯数字兜底
    m = re.search(r"([\d.]+)", raw)
    if m:
        return {"price": Decimal(m.group(1)), "currency": "unknown"}
    return {"price": 0, "currency": "unknown"}
```

## 第三步：中文编码修复

跨境数据采集常遇到编码问题，特别是繁体中文和 GBK 编码的数据：

```python
def fix_encoding(text: str) -> str:
    """修复常见编码问题"""
    if not text:
        return text
    
    # 常见乱码替换
    fixes = {
        "çš„": "的", "æ˜¯": "是", "äº†": "了",
        "ä¸": "不", "æœ‰": "有", "åœ¨": "在",
    }
    for bad, good in fixes.items():
        text = text.replace(bad, good)
    
    return text

def clean_encoding(item: Dict) -> Dict:
    """递归修复所有字符串字段的编码"""
    result = {}
    for k, v in item.items():
        if isinstance(v, str):
            result[k] = fix_encoding(v)
        elif isinstance(v, dict):
            result[k] = clean_encoding(v)
        elif isinstance(v, list):
            result[k] = [clean_encoding({"": x})[""] if isinstance(x, str) else x for x in v]
        else:
            result[k] = v
    return result
```

## 完整管线示例

```python
# 组装管线
pipeline = CleaningPipeline()
pipeline.add_step("字段映射", clean_fields)
pipeline.add_step("编码修复", clean_encoding)

def standardize_prices(item: Dict) -> Dict:
    if "price" in item and isinstance(item["price"], str):
        item.update(normalize_price(item["price"]))
    return item
pipeline.add_step("价格标准化", standardize_prices)

# 模拟多平台数据
raw_data = [
    {"title": "iPhone 15", "price": "¥6,999", "name": "iPhone 15"},
    {"title": "AirPods Pro", "sale_price": "$249.99", "img": "airpods.jpg"},
    {"title": "çš„å•†å“", "price": "9,800円"},  # 模拟乱码
]

cleaned = pipeline.run(raw_data)
# 输出标准化后的 JSON 数据
```

## 与业务系统的集成

数据清洗管线不是孤立存在的，它通常作为数据采集模块的前置环节。在 taocarts 这类代购系统中，从 1688、淘宝、日本乐天等多平台采集商品时，每批数据入库前都会经过类似的清洗流程，确保后续的比价、推荐、库存同步等功能基于干净的数据运行。

## 总结

数据清洗是数据工程中最耗时但最必要的环节。以上设计模式的核心思想是：每个清洗步骤只做一件事，通过管道串联，方便排查和扩展。

跨境商品数据采集面临格式不统一、字段缺失、编码混乱等问题。不同平台的商品数据，有的用 JSON，有的是 HTML 片段，有的直接是纯文本描述。不做清洗的话，这些数据没法直接入库分析。

跨境商品数据清洗管线：从多平台采集到结构化存储

构建数据清洗管线，将跨境商品从多平台采集的脏数据转化为可用结构

php后端

数据库

大数据

后端

跨境商品数据ETL管线设计：解决多平台数据格式混乱问题，通过解析层、清洗层、标准化层、校验层实现数据统一。包含字段映射、价格转换、编码修复等核心步骤，采用管道模式串联处理，输出结构化数据。适用于1688、淘宝、乐天等平台采集场景，提升数据质量。

结构化数据

数据清洗

代码解释

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

跨境商品数据清洗管线：从多平台采集到结构化存储-腾讯云开发者社区-腾讯云

跨境商品数据清洗管线：从多平台采集到结构化存储

跨境商品数据清洗管线：从多平台采集到结构化存储

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐