如何清洗15 Lkh未过滤数据集并经Pandas校验后上传至MySQL

针对15Lkh（150万）条未过滤数据集的规整、校验与入库需求，以下是可直接落地的流程与代码实现：

1. 初始数据规整：去重与缺失值清理 #

先完成基础的数据清洗，移除重复项与无效缺失值：

import pandas as pd

# 读取大规模数据集（low_memory=False避免类型推断警告，大文件可加chunksize分块处理）
df = pd.read_csv("raw_data.csv", low_memory=False)

# 移除重复行（可选：指定subset参数按关键列去重）
df = df.drop_duplicates(keep='first')

# 删除全为空的无效行
df = df.dropna(how='all')

# 针对核心业务列（如用户ID、交易日期）删除缺失行
critical_cols = ['user_id', 'transaction_date', 'amount']
df = df.dropna(subset=critical_cols)

# 非核心列缺失值填充（按需选择策略）
# 数值列用中位数填充
df['age'] = df['age'].fillna(df['age'].median())
# 分类列用众数填充
df['product_category'] = df['product_category'].fillna(df['product_category'].mode()[0])

2. Pandas二次处理与双重校验 #

完成基础清洗后，进行数据格式修正与业务规则校验，确保数据合规：

2.1 二次数据标准化处理 #

# 修正日期格式，过滤无效日期
df['transaction_date'] = pd.to_datetime(df['transaction_date'], errors='coerce')
df = df.dropna(subset=['transaction_date'])

# 统一文本格式（去除空格、转小写）
df['product_name'] = df['product_name'].str.strip().str.lower()

# 业务规则过滤（如交易金额不能为负）
df = df[df['amount'] > 0]

2.2 双重校验确保数据完整性 #

# 校验1：数据量核对
print(f"初始数据量：1500000条")
print(f"清洗后数据量：{len(df)}条")
print(f"累计清理行数：{1500000 - len(df)}条")

# 校验2：核心列无缺失
null_check = df[critical_cols].isnull().sum()
if null_check.any():
    raise ValueError(f"核心列仍存在缺失：{null_check}")

# 校验3：业务规则合规性
invalid_amount = len(df[df['amount'] <= 0])
if invalid_amount > 0:
    raise ValueError(f"发现{invalid_amount}条无效金额数据")

3. 规整存储至MySQL数据库 #

使用SQLAlchemy实现高效入库，并完成最终校验：

3.1 配置数据库连接 #

from sqlalchemy import create_engine

# 替换为你的实际数据库配置
db_settings = {
    'user': 'db_username',
    'password': 'db_password',
    'host': 'localhost',
    'port': 3306,
    'db': 'your_database'
}

# 创建数据库连接引擎
engine = create_engine(
    f"mysql+mysqlconnector://{db_settings['user']}:{db_settings['password']}@{db_settings['host']}:{db_settings['port']}/{db_settings['db']}"
)

3.2 数据入库与最终校验 #

# 分块上传数据（chunksize避免内存溢出，if_exists按需选'replace'或'append'）
df.to_sql(
    name='cleaned_transactions',  # 目标表名
    con=engine,
    if_exists='replace',
    index=False,
    chunksize=10000
)

# 入库后校验：核对本地与数据库数据量
with engine.connect() as conn:
    db_row_count = conn.execute("SELECT COUNT(*) FROM cleaned_transactions").scalar()

if db_row_count != len(df):
    raise ValueError(f"入库数据不一致：本地{len(df)}条，数据库{db_row_count}条")
else:
    print("数据成功入库，所有校验通过！")

关键注意事项 #

• 处理超大规模数据时，优先用chunksize分块读取处理，避免内存耗尽
• 缺失值处理需结合业务场景，不要盲目删除或填充
• 数据库凭证建议通过环境变量管理，禁止硬编码
• 入库前建议提前创建表结构并添加索引，提升后续查询效率

内容的提问来源于stack exchange，提问作者Prashant Gaur