# Pandas + SQLite3 数据清洗实战：从脏数据到干净数据库的完整流程 每次打开满是缺失值、格式混乱的CSV文件时，你是否也感到头皮发麻？上周我接手了一个电商用户行为数据集，37%的年龄字段是"unknown"，15%的邮箱带着乱码，还有大量前后不一致的日期格式。这种"脏数据"就像未加工的食材，直接下锅只会毁掉整道菜。本文将带你用Pandas和SQLite3打造一套数据清洗流水线，把原始数据变成结构化的数据库美味。 ## 1. 环境配置与数据初探 工欲善其事，必先利其器。这套组合拳只需要两个核心工具： ```bash pip install pandas numpy ``` SQLite3作为Python内置库无需额外安装，但建议同时安装NumPy以获得更好的数值处理能力。新建一个Jupyter Notebook或Python脚本，导入以下工具包： ```python import pandas as pd import sqlite3 from pathlib import Path import re ``` **典型脏数据症状检查表**： | 问题类型 | 表现形式 | 危害性 | |----------------|---------------------------|--------| | 缺失值 | NaN/Null/空字符串 | ★★★★ | | 格式不一致 | 日期混用20230101和Jan-2023 | ★★★☆ | | 异常值 | 年龄=300 | ★★★★ | | 重复数据 | 完全相同的多行记录 | ★★☆☆ | | 字符串污染 | "北京 "带空格/"北京" | ★★☆☆ | > 提示：先用`df.info()`和`df.describe()`快速扫描数据概况，再用`df.sample(5)`随机抽查记录，比直接看`head()`更容易发现隐藏问题。 ## 2. 结构化清洗流程设计 ### 2.1 缺失值处理策略矩阵 不是所有缺失值都需要填充或删除，根据业务场景选择合适策略： ```python # 删除缺失超过30%的列 threshold = len(df) * 0.3 df = df.dropna(thresh=threshold, axis=1) # 分类型填充 fill_strategy = { 'age': df['age'].median(), # 数值用中位数 'email': 'unknown', # 分类用特定标记 'last_login': pd.Timestamp.now() # 时间用当前值 } df = df.fillna(fill_strategy) ``` ### 2.2 正则表达式清洗工坊 杂乱字符串是数据清洗的顽固污渍，正则表达式就是最强去污剂： ```python def clean_string(col): # 去除特殊字符 col = col.apply(lambda x: re.sub(r'[^\w\s-]', '', str(x))) # 统一手机号格式 col = col.apply(lambda x: re.sub(r'(\d{3})(\d{4})(\d{4})', r'\1-\2-\3', x)) # 英文首字母大写 col = col.str.title() return col df['name'] = clean_string(df['name']) df['phone'] = clean_string(df['phone']) ``` **常见正则表达式清洗场景**： 1. 邮箱验证：`^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$` 2. URL提取：`https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+` 3. 中文提取：`[\u4e00-\u9fa5]+` ## 3. 数据库交互最佳实践 ### 3.1 智能类型推断写入 Pandas的`to_sql()`默认将所有字段存储为TEXT类型，这会导致后续查询性能低下。解决方案是提前建立类型映射： ```python def pandas_to_sqlite_type(dtype): if pd.api.types.is_integer_dtype(dtype): return 'INTEGER' elif pd.api.types.is_float_dtype(dtype): return 'REAL' elif pd.api.types.is_datetime64_dtype(dtype): return 'TIMESTAMP' else: return 'TEXT' # 自动生成类型字典 dtype_dict = {col: pandas_to_sqlite_type(df[col].dtype) for col in df.columns} ``` ### 3.2 事务处理与性能优化 大批量写入时，这些技巧可以提升10倍以上性能： ```python with sqlite3.connect('cleaned.db') as conn: # 禁用自动提交 conn.execute("PRAGMA synchronous = OFF") conn.execute("PRAGMA journal_mode = MEMORY") # 分块写入 chunk_size = 10000 for i in range(0, len(df), chunk_size): chunk = df.iloc[i:i + chunk_size] chunk.to_sql('users', conn, if_exists='append', index=False, dtype=dtype_dict) # 建立索引(写入完成后再创建) conn.execute("CREATE INDEX idx_user_email ON users(email)") ``` > 注意：在金融等关键领域，不建议禁用同步设置，以防电源故障导致数据损坏。 ## 4. 实战：电商用户数据清洗案例 假设我们有一个`dirty_users.csv`文件，包含以下问题数据： ``` id,username,age,join_date,purchase_history 1, john_doe , 32, 2022-01-15, "[{'id':101,'amount':199}]" 2, alice-smith, twenty-five, 15/01/2023, "[]" 3, BOB, 45, 2023-02-30, "[{'id':102,'amount':'五十元'}]" ``` **分步解决方案**： 1. **元数据清洗**： ```python df = pd.read_csv('dirty_users.csv', converters={ 'purchase_history': eval # 自动解析JSON字符串 }) # 统一日期格式 df['join_date'] = pd.to_datetime(df['join_date'], errors='coerce', dayfirst=True) # 年龄字段标准化 df['age'] = pd.to_numeric(df['age'], errors='coerce') df['age'] = df['age'].fillna(df['age'].median()).astype(int) ``` 2. **嵌套数据处理**： ```python # 展开JSON数组 purchase_df = pd.json_normalize( df['purchase_history'].explode() ).groupby(level=0).sum() df = df.join(purchase_df['amount'].rename('total_purchase')) ``` 3. **最终数据验证**： ```python # 创建数据质量报告 validation_rules = { 'age_range': (df['age'] >= 18) & (df['age'] <= 100), 'valid_date': df['join_date'].notna(), 'positive_purchase': df['total_purchase'] >= 0 } for name, rule in validation_rules.items(): print(f"{name} pass rate: {rule.mean():.1%}") ``` ## 5. 自动化清洗流水线搭建 将上述流程封装成可复用的管道： ```python class DataCleaningPipeline: def __init__(self, db_path='cleaned.db'): self.db_path = Path(db_path) if self.db_path.exists(): self.db_path.unlink() def clean_strings(self, df): # 字符串标准化处理 df = df.apply(lambda col: col.str.strip() if col.dtype == 'object' else col) return df def process_file(self, file_path, table_name): df = pd.read_csv(file_path) df = self.clean_strings(df) with sqlite3.connect(self.db_path) as conn: df.to_sql(table_name, conn, index=False) return f"Data saved to {table_name}" # 使用示例 pipeline = DataCleaningPipeline() pipeline.process_file('dirty_data.csv', 'clean_users') ``` **流水线扩展功能建议**： 1. 添加数据质量报告生成 2. 支持Excel/JSON等多种输入格式 3. 集成邮件通知功能 4. 添加数据血缘追踪 清洗数据就像考古修复，需要耐心和合适的工具。最近处理的一个物流数据集让我深刻体会到：80%的数据分析时间其实花在了数据准备上。当看到杂乱的数据最终变成整洁的数据库表时，那种成就感不亚于解决一个复杂算法问题。记住，好的数据清洗流程应该像隐形眼镜护理液——平时感觉不到它的存在，但一旦缺少就会非常难受。