1. 当数据撞上MongoDB：那些年我们踩过的坑

凌晨三点，程序员的咖啡杯见了底，控制台突然弹出红色警告："BSON conversion error"。这是每位MongoDB使用者都可能遇到的经典场景——数据格式不符合预期就像快递小哥送错了包裹，不仅耽误业务处理，还可能引发系统崩溃。

举个真实案例： 某电商平台迁移用户数据时，发现订单时间字段存在多种格式：

{
    "order_id": "20230815-001",
    "create_time": "2023-08-15 14:30",  # 字符串格式
    "pay_time": 1692070200.0,          # Unix时间戳
    "delivery_time": ISODate("2023-08-16T08:00:00Z")  # MongoDB原生日期类型
}

当这些数据通过PyMongo批量导入时，MongoDB会因日期类型不统一而拒绝部分文档的写入。这种"混合格式症候群"正是数据清洗需要解决的首要问题。

2. 数据清洗三板斧：过滤、转换、校验

2.1 数据类型标准化（Python + PyMongo方案）

from datetime import datetime
import pandas as pd

def time_converter(value):
    """万能时间格式转换器"""
    try:
        # 尝试解析字符串格式
        if isinstance(value, str):
            return datetime.strptime(value, "%Y-%m-%d %H:%M")
        # 处理Unix时间戳
        elif isinstance(value, (int, float)):
            return datetime.fromtimestamp(value)
        # 兼容已有日期类型
        elif isinstance(value, datetime):
            return value
    except Exception as e:
        print(f"时间转换异常：{value} -> {str(e)}")
        return None

# 使用Pandas预处理数据
df = pd.read_csv('orders.csv')
df['create_time'] = df['create_time'].apply(time_converter)
df['pay_time'] = df['pay_time'].apply(time_converter)

# 空值处理
df = df.dropna(subset=['create_time'])  # 删除关键时间缺失的记录
df.fillna({'delivery_time': '未发货'}, inplace=True)  # 填充特殊标记

# 转换为字典列表进行插入
data_dict = df.to_dict('records')
collection.insert_many(data_dict)

注释说明：

  • 通过分层转换策略处理多种时间格式
  • 使用Pandas进行批量化处理提升效率
  • 采用防御性编程处理异常值

2.2 嵌套结构拆解技巧

遇到多层嵌套的JSON数据时：

# 原始用户行为数据
user_actions = {
    "user_id": 1001,
    "activities": [
        {"type": "login", "time": "2023-08-15 09:00"},
        {"type": "search", "keywords": "MongoDB教程"}
    ]
}

# 结构扁平化处理
processed = []
for action in user_actions['activities']:
    base = {
        "user_id": user_actions['user_id'],
        "action_time": time_converter(action.get('time'))
    }
    base.update({k:v for k,v in action.items() if k != 'time'})
    processed.append(base)

collection.insert_many(processed)  # 转换为更适合查询的文档结构

3. 技术选型：PyMongo的攻守之道

3.1 适用场景

  • 中小型数据迁移（GB级以下）
  • 需要复杂业务逻辑的数据转换
  • 开发测试环境的数据准备

3.2 性能表现

优势：

  • 开发效率高：直接使用Python生态工具链
  • 灵活性好：支持自定义清洗逻辑
  • 调试方便：可在Jupyter等环境中逐步验证

劣势：

  • 内存消耗大：Pandas处理超大数据时易崩溃
  • 执行速度慢：相比mongoimport工具性能下降约40%
  • 缺乏断点续传：批量插入失败需全量重试

4. 避坑指南：从新手到专家的必经之路

4.1 预处理三原则

  • 格式验证先行：使用jsonchema校验文档结构
  • 分批处理策略：每5000条提交一次批量写入
  • 版本化管理：保留原始数据和清洗脚本的对应版本

4.2 性能优化技巧

# 慢速写法（逐条插入）
for doc in data_list:
    collection.insert_one(doc)

# 高速写法（批量插入+有序配置）
collection.insert_many(
    data_list, 
    ordered=False,  # 允许并行处理
    bypass_document_validation=True  # 跳过重复校验
)

5. 战场经验总结

经过数十次数据迁移的洗礼，我们得出三个黄金法则： 1）数据清洗应该像洋葱一样分层处理，每层只解决特定类型问题 2）永远保留原始数据的"快照"，清洗过程必须可追溯 3）在开发环境建立完整的异常样本库，持续完善清洗规则

6.最终建议

对于TB级数据迁移，建议采用mongoimport工具直接导入，而复杂业务场景下的数据清洗，PyMongo+Python生态仍是目前的最佳选择。记住，好的数据清洗就像给数据库做SPA——既要彻底清洁，又要保持数据活性。