第02课：ETL管道设计

🔧 ETL管道设计

ETL（Extract-Transform-Load）是数据工程最基础的模式。一个设计良好的ETL管道是可靠数据流的基石。

🐍 Python实战：核心实战

# 构建可扩展的ETL管道框架
import time, random
from datetime import datetime
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class PipelineContext:
    data: list = field(default_factory=list)
    metadata: dict = field(default_factory=dict)
    metrics: dict = field(default_factory=dict)

class PipelineStep(ABC):
    def __init__(self, name): self.name = name
    @abstractmethod
    def execute(self, context): pass

class ExtractStep(PipelineStep):
    def __init__(self, name, count=200):
        super().__init__(name)
        self.count = count
    def execute(self, context):
        records = []
        for i in range(self.count):
            records.append({'id':i+1,'value':round(random.uniform(10,1000),2),
                'category':random.choice(['A','B','C','D']),
                'status':random.choice(['active','inactive','pending'])})
        context.data.extend(records)
        context.metrics['extracted'] = len(records)
        return context

class CleanStep(PipelineStep):
    def __init__(self, name, rules):
        super().__init__(name)
        self.rules = rules
    def execute(self, context):
        cleaned = []
        removed = defaultdict(int)
        for r in context.data:
            skip = False
            for fld, rule in self.rules.items():
                if fld in r:
                    val = r[fld]
                    if rule.get('min') is not None and isinstance(val,(int,float)) and val < rule['min']:
                        removed[f'{fld}_low'] += 1; skip = True; break
                    if rule.get('valid') and val not in rule['valid']:
                        removed[f'{fld}_invalid'] += 1; skip = True; break
            if not skip: cleaned.append(r)
        context.data = cleaned
        context.metrics['cleaned'] = len(cleaned)
        context.metadata['removed'] = dict(removed)
        return context

class AggregateStep(PipelineStep):
    def __init__(self, name, group_by, agg_field):
        super().__init__(name)
        self.group_by = group_by
        self.agg_field = agg_field
    def execute(self, context):
        groups = defaultdict(list)
        for r in context.data:
            groups[r.get(self.group_by,'unknown')].append(r.get(self.agg_field,0))
        context.data = [{self.group_by:k,'total':round(sum(v),2),'count':len(v),'avg':round(sum(v)/len(v),2)}
                       for k,v in groups.items()]
        context.metrics['groups'] = len(context.data)
        return context

class ETLPipeline:
    def __init__(self, name):
        self.name = name
        self.steps = []
        self.context = PipelineContext()
    def add_step(self, step): self.steps.append(step); return self
    def run(self):
        print(f"启动管道: {self.name} ({len(self.steps)}步骤)")
        for step in self.steps:
            start = time.perf_counter()
            inp = len(self.context.data)
            self.context = step.execute(self.context)
            out = len(self.context.data)
            ms = (time.perf_counter()-start)*1000
            print(f"  ✅ {step.name}: {inp}→{out} ({ms:.1f}ms)")
        return self.context

pipeline = ETLPipeline("订单处理管道")
pipeline.add_step(ExtractStep("提取订单", 200))
pipeline.add_step(CleanStep("数据清洗", {'value':{'min':0},'status':{'valid':['active','inactive','pending']}}))
pipeline.add_step(AggregateStep("分类聚合", 'category', 'value'))
result = pipeline.run()
print(f"最终: {len(result.data)}组")
for r in result.data[:4]:
    print(f"  {r}")
print("✅ 验证通过 - ETL管道框架运行正常")

📐 管道设计模式

🐍 管道可靠性：重试、幂等、断点续传

# 管道可靠性：重试、幂等、断点续传
import time, random
from functools import wraps

def retry(max_attempts=3, backoff_base=2):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts+1):
                try: return fn(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts: raise
                    wait = backoff_base ** attempt + random.uniform(0,1)
                    print(f"  第{attempt}次失败，{wait:.1f}s后重试: {e}")
                    time.sleep(wait)
        return wrapper
    return decorator

class IdempotentLoader:
    def __init__(self): self.loaded_ids = set(); self.dedup = 0
    def load(self, records, id_field='id'):
        new = []
        for r in records:
            rid = r.get(id_field)
            if rid not in self.loaded_ids: self.loaded_ids.add(rid); new.append(r)
            else: self.dedup += 1
        print(f"  幂等加载: {len(records)}输入, {len(new)}新增, {self.dedup}去重")
        return new

class Checkpoint:
    def __init__(self): self.done = set(); self.state = {}
    def mark(self, name, state=None): self.done.add(name); self.state[name] = state
    def is_done(self, name): return name in self.done

@retry(max_attempts=3, backoff_base=1)
def unreliable_api():
    if random.random() < 0.5: raise ConnectionError("API超时")
    return [{"id":1,"data":"ok"}]

print("可靠性机制测试")
print("1. 重试:")
try: print(f"  结果: {unreliable_api()}")
except: print("  最终失败")
print("2. 幂等:")
loader = IdempotentLoader()
loader.load([{'id':i,'v':i*10} for i in range(1,11)])
loader.load([{'id':i,'v':i*10} for i in range(1,6)])
print("3. 断点续传:")
ckpt = Checkpoint()
ckpt.mark('extract', {'offset':1000})
print(f"  extract完成: {ckpt.is_done('extract')}")
print("✅ 验证通过 - 可靠性机制运行正常")

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

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

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

📚 延伸阅读与参考