第04课：数据转换-清洗与标准化

🔄 数据转换-清洗与标准化

数据转换是ETL核心环节，将脏数据转化为高质量的分析就绪数据。据统计数据工程师60-80%时间花在清洗转换上。

🐍 Python实战：核心实战

# 数据清洗框架
import re, random
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass

class DataCleaner:
    def __init__(self): self.reports = []
    def handle_missing(self, records, field, strategy='drop', fill=None):
        orig = len(records)
        missing = sum(1 for r in records if r.get(field) is None or r.get(field) == '')
        if strategy == 'drop':
            result = [r for r in records if r.get(field) is not None and r.get(field) != '']
        elif strategy == 'fill':
            result = []
            for r in records:
                r = r.copy()
                if r.get(field) is None or r.get(field) == '': r[field] = fill
                result.append(r)
        elif strategy == 'mean_fill':
            vals = [r[field] for r in records if r.get(field) is not None and isinstance(r[field],(int,float))]
            mean = sum(vals)/len(vals) if vals else 0
            result = []
            for r in records:
                r = r.copy()
                if r.get(field) is None or r.get(field) == '': r[field] = round(mean,2)
                result.append(r)
        else: result = records
        self.reports.append(f"[missing] {field}: {orig}→{len(result)} ({missing}空值,{strategy})")
        return result
    def deduplicate(self, records, key_fields, strategy='last'):
        orig = len(records)
        seen = {}
        for r in records:
            key = tuple(r.get(k) for k in key_fields)
            if key in seen:
                if strategy == 'merge':
                    for k,v in r.items():
                        if v is not None and v != '' and (seen[key].get(k) is None or seen[key].get(k) == ''):
                            seen[key][k] = v
                elif strategy == 'last': seen[key] = r.copy()
            else: seen[key] = r.copy()
        result = list(seen.values())
        self.reports.append(f"[dup] {','.join(key_fields)}: {orig}→{len(result)} ({orig-len(result)}重复)")
        return result
    def detect_outliers(self, records, field, method='iqr', threshold=1.5):
        values = [r[field] for r in records if field in r and isinstance(r[field],(int,float))]
        if not values: return records, []
        sv = sorted(values); n = len(sv)
        q1, q3 = sv[n//4], sv[3*n//4]; iqr = q3-q1
        lower, upper = q1-threshold*iqr, q3+threshold*iqr
        outliers = [i for i,r in enumerate(records) if field in r and isinstance(r[field],(int,float)) and (r[field]upper)]
        self.reports.append(f"[outlier] {field}({method}): {len(outliers)}异常, 范围[{lower:.1f},{upper:.1f}]")
        return records, outliers

class Standardizer:
    def __init__(self): self.log = []
    def normalize_date(self, records, field, out_fmt='%Y-%m-%d'):
        pats = [(r'\d{4}-\d{2}-\d{2}','%Y-%m-%d'),(r'\d{2}/\d{2}/\d{4}','%m/%d/%Y'),
                (r'\d{4}/\d{2}/\d{2}','%Y/%m/%d'),(r'\d{8}','%Y%m%d')]
        converted = 0
        result = []
        for r in records:
            r = r.copy(); val = r.get(field,'')
            if isinstance(val,str) and val:
                for pat, fmt in pats:
                    if re.match(f'^{pat}$', val):
                        try:
                            dt = datetime.strptime(val, fmt)
                            r[field] = dt.strftime(out_fmt)
                            converted += 1; break
                        except ValueError: pass
            result.append(r)
        self.log.append(f"日期标准化 {field}: {converted}条转换")
        return result
    def scale_numeric(self, records, field, method='minmax'):
        values = [r[field] for r in records if field in r and isinstance(r[field],(int,float))]
        if not values: return records
        min_v, max_v = min(values), max(values)
        result = []
        for r in records:
            r = r.copy()
            if field in r and isinstance(r[field],(int,float)) and max_v > min_v:
                r[f'{field}_scaled'] = round((r[field]-min_v)/(max_v-min_v), 4)
            result.append(r)
        self.log.append(f"数值标准化 {field}({method}): [{min_v:.1f},{max_v:.1f}]")
        return result
    def encode_categorical(self, records, field, method='label'):
        cats = sorted(set(r[field] for r in records if field in r and r[field] is not None))
        result = []
        for r in records:
            r = r.copy()
            val = r.get(field)
            if method == 'label': r[f'{field}_enc'] = cats.index(val) if val in cats else -1
            result.append(r)
        self.log.append(f"类别编码 {field}: {len(cats)}类")
        return result

random.seed(42)
dirty = []
for i in range(500):
    dirty.append({'id':i+1, 'name':random.choice([f'用户{i+1}','',None]),
        'email':random.choice([f'user{i}@test.com','']),
        'age':random.choice([random.randint(18,80),None,-1,999]),
        'salary':random.choice([round(random.uniform(3000,50000),2),None]),
        'join_date':random.choice(['2024-01-15','01/15/2024','20240115','']),
        'dept':random.choice(['Engineering','engineering','Sales',None])})
dirty.extend(dirty[:20])

cl = DataCleaner()
std = Standardizer()
data = cl.handle_missing(dirty, 'email', 'fill', 'unknown@null.com')
data = cl.handle_missing(data, 'age', 'mean_fill')
data = cl.deduplicate(data, ['id'], 'merge')
_, age_out = cl.detect_outliers(data, 'age', 'iqr')
data = std.normalize_date(data, 'join_date')
data = std.scale_numeric(data, 'salary', 'minmax')
data = std.encode_categorical(data, 'dept', 'label')
print(f"清洗: {len(dirty)}→{len(data)}条")
for r in cl.reports: print(f"  {r}")
for r in std.log: print(f"  {r}")
print("✅ 验证通过 - 数据清洗框架运行正常")

🐍 转换性能基准

import time, random
sizes = [1000, 5000, 10000, 50000]
for size in sizes:
    data = [{'id':i,'value':random.uniform(0,1000),'cat':random.choice('ABCD')} for i in range(size)]
    start = time.perf_counter()
    _ = [r for r in data if r['value'] > 100]
    t1 = (time.perf_counter()-start)*1000
    start = time.perf_counter()
    _ = sorted(data, key=lambda r: r['value'])
    t2 = (time.perf_counter()-start)*1000
    print(f"{size:>8,}条: 过滤{t1:>8.2f}ms, 排序{t2:>8.2f}ms")
print("✅ 验证通过 - 性能基准完成")

📖 原理深入：ETL设计核心原则

🏆 最佳实践：生产环境ETL检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考