1. 数据清洗：从“脏数据”到“可复现资产”的蜕变

数据清洗是每个数据分析师和数据科学家职业生涯中绕不开的必修课。记得我第一次处理电商用户行为数据时，面对40%的缺失值和五花八门的日期格式，才真正理解为什么业内常说"数据科学家80%的时间都在清洗数据"。数据清洗不是简单的数据"美容"，而是将原始数据转化为可信赖分析基础的关键过程。

当我们谈论"脏数据"时，通常指数据中存在以下几种典型问题：

• 缺失值：用户未填写的表单字段、传感器中断的记录
• 重复数据：系统重复录入、爬虫多次抓取的相同内容
• 格式混乱：日期有"2023/01/01"也有"01-Jan-2023"
• 异常值：超出合理范围的数值（如年龄200岁）
• 不一致性：同一实体在不同系统的不同编码（如"北京"vs"北京市"）

这些问题如果不处理，会导致分析结果严重失真。我曾见过一个案例，由于未处理重复交易记录，导致销售预测虚高了23%。数据清洗的本质，是通过系统化的方法识别和修正这些问题，使数据达到"分析就绪"状态。

2. 缺失值处理：不只是简单填充或删除

2.1 缺失值类型识别

处理缺失值的第一步是理解其产生机制。根据统计学的分类，缺失值可分为：

1. 完全随机缺失(MCAR)：缺失与任何变量无关（如服务器随机丢包）
2. 随机缺失(MAR)：缺失与已观测变量相关（如女性更可能不填写收入）
3. 非随机缺失(MNAR)：缺失与缺失值本身相关（如高收入者不愿披露收入）

识别缺失类型对选择处理方法至关重要。我常用的诊断方法是：

import missingno as msno msno.matrix(df) # 可视化缺失模式 msno.heatmap(df) # 检查变量间缺失相关性

2.2 处理方法选择与实践

删除法

• 列表删除：直接删除含缺失值的行

  • 适用场景：缺失比例<5%且为MCAR
  • Python实现：df.dropna(how='any')
  • 风险：可能引入偏差（如删除所有高收入缺失记录）

• 列删除：删除缺失率过高的列

  • 阈值建议：通常>40%缺失的列考虑删除
  • 代码：df.drop(columns=df.columns[df.isnull().mean() > 0.4])

填充法

• 单值填充：

  # 均值填充（对正态分布数据） df['income'].fillna(df['income'].mean(), inplace=True) # 中位数填充（对偏态分布） df['age'].fillna(df['age'].median(), inplace=True) # 众数填充（分类变量） df['gender'].fillna(df['gender'].mode()[0], inplace=True)

• 预测填充（更高级的方法）：

  from sklearn.experimental import IterativeImputer imputer = IterativeImputer(max_iter=10) df_imputed = pd.DataFrame(imputer.fit_transform(df), columns=df.columns)

重要提示：任何填充方法都会引入偏差，必须记录填充方法并在报告中说明，否则可能误导分析结论。

特殊标记法

对于MNAR情况，建议将缺失本身作为信息保留：

df['income_missing'] = df['income'].isnull().astype(int)

3. 数据去重：不仅仅是drop_duplicates()

3.1 精确去重技术

基础的去重方法在Pandas中很简单：

# 基于所有列完全匹配 df.drop_duplicates(inplace=True) # 基于特定列 df.drop_duplicates(subset=['user_id', 'order_date'], keep='last')

但实际业务中常遇到更复杂情况：

3.2 模糊去重实战

处理姓名、地址等文本数据时，需要模糊匹配：

from fuzzywuzzy import fuzz def fuzzy_match(str1, str2): return fuzz.token_set_ratio(str1, str2) > 80 # 示例：合并相似公司名称 companies = ['Microsoft Corp', 'Microsoft Corporation', 'MSFT']

3.3 基于业务规则的去重

电商订单去重的典型例子：

# 保留同一用户同一商品最近30天内的最高金额订单 window = Window.partitionBy('user_id', 'product_id').orderBy(F.desc('amount')) df = df.withColumn('rank', F.rank().over(window)) df_filtered = df.filter(df.rank == 1).drop('rank')

3.4 大数据去重优化

当数据量达到TB级时，需要特殊技术：

• 布隆过滤器：快速判断元素是否存在
• MinHash：近似检测相似文档
• StarRocks的Bitmap去重：适合超大规模数据集

-- StarRocks中使用Bitmap精确去重 SELECT user_id, COUNT(DISTINCT item_id) AS unique_items FROM user_behavior GROUP BY user_id;

4. 数据类型转换：避免隐式转换的陷阱

4.1 常见类型转换场景

字符串转日期

# 处理多种日期格式 def parse_date(date_str): for fmt in ('%Y-%m-%d', '%d/%m/%Y', '%b %d %Y'): try: return datetime.strptime(date_str, fmt) except ValueError: continue return pd.NaT # 无法解析的返回缺失 df['date'] = df['date_str'].apply(parse_date)

分类数据编码

# 有序分类变量 size_map = {'XS':1, 'S':2, 'M':3, 'L':4, 'XL':5} df['size_code'] = df['size'].map(size_map) # 无序分类变量（One-Hot） df = pd.get_dummies(df, columns=['color'], prefix='', prefix_sep='')

数值类型处理

# 处理货币字符串 df['price'] = df['price'].str.replace('$','').str.replace(',','').astype(float) # 处理百分比 df['discount'] = df['discount'].str.replace('%','').astype(float) / 100

4.2 类型转换中的常见坑

1. 隐式转换风险：

   # 危险操作：混合类型的列 df = pd.DataFrame({'col': [1, 2, '3', 4]}) df['col'].mean() # 抛出TypeError

2. 时区问题：

   # 最佳实践：始终使用UTC时间戳 df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)

3. 大整数精度丢失：

   # 使用Python原生int类型处理大整数 df['big_id'] = df['big_id'].astype(object).apply(lambda x: int(x))

5. 工业级数据清洗流水线设计

5.1 可复现的清洗流程

我推荐的清洗流水线结构：

raw_data/ ├── input/ # 原始数据 ├── processed/ # 清洗后数据 ├── scripts/ │ ├── 01_clean_missing.py │ ├── 02_deduplicate.py │ └── 03_transform.py └── config/ └── cleaning_rules.yaml # 清洗规则配置

使用Makefile自动化流程：

.PHONY: clean_all clean_all: python scripts/01_clean_missing.py python scripts/02_deduplicate.py python scripts/03_transform.py %.csv: %.raw.csv python scripts/01_clean_missing.py $< $@

5.2 数据质量验证

清洗后必须进行质量检查：

def validate_data(df): # 检查缺失率 assert df.isnull().mean().max() < 0.05, "缺失值超过阈值" # 检查ID唯一性 assert df['user_id'].nunique() == len(df), "user_id不唯一" # 检查数值范围 assert df['age'].between(18, 100).all(), "年龄超出合理范围" # 检查日期单调性 assert df['event_time'].is_monotonic_increasing, "时间非单调递增"

5.3 性能优化技巧

处理大型数据集时：

• 分块处理：

  chunk_iter = pd.read_csv('large.csv', chunksize=100000) for chunk in chunk_iter: process(chunk)

• 并行处理：

  from multiprocessing import Pool def parallel_clean(df): with Pool(4) as p: results = p.map(clean_function, np.array_split(df, 4)) return pd.concat(results)

• 数据库内清洗：

  -- 直接在数据库中执行复杂清洗 CREATE TABLE clean_data AS WITH deduped AS ( SELECT *, ROW_NUMBER() OVER(PARTITION BY user_id ORDER BY event_time DESC) AS rn FROM raw_events ) SELECT * FROM deduped WHERE rn = 1;

数据清洗不是一次性的工作，而是需要持续优化的过程。每次分析新数据集时，我都会记录遇到的特殊情况和处理方式，逐渐积累成自己的"数据清洗模式库"。这个习惯让我在面对新的"脏数据"挑战时能够快速找到合适的解决方案。