ETL基础

第05课：数据加载-批量与增量

数据工程实战课程 · ETL基础阶段

📤 数据加载-批量与增量

数据加载是ETL最后一步，加载策略的选择直接影响数据时效性、系统稳定性和存储效率。

加载策略对比

策略	原理	场景
全量覆盖	TRUNCATE+INSERT	小维度表
增量Upsert	ON CONFLICT	状态表
SCD Type2	版本追踪	维度表

🐍 Python实战：核心实战

# 数据加载框架（SQLite实机）
import sqlite3, time, random
from datetime import datetime
from dataclasses import dataclass

conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute('CREATE TABLE dim_customer (id INTEGER PRIMARY KEY, name TEXT, tier TEXT, is_current INT DEFAULT 1, valid_from TEXT, valid_to TEXT, scd_version INT DEFAULT 1)')
c.execute('CREATE TABLE fact_orders (id INTEGER PRIMARY KEY AUTOINCREMENT, order_key TEXT UNIQUE, customer_id INT, amount REAL, status TEXT)')
conn.commit()

class DataLoader:
    def __init__(self, conn):
        self.conn = conn
        self.history = []
    def full_replace(self, table, records):
        start = time.perf_counter()
        cur = self.conn.cursor()
        temp = f"{table}_new"
        cur.execute(f"CREATE TABLE {temp} AS SELECT * FROM {table} WHERE 1=0")
        inserted = 0
        for r in records:
            cols, vals = ', '.join(r.keys()), ', '.join(['?' for _ in r])
            try: cur.execute(f"INSERT INTO {temp} ({cols}) VALUES ({vals})", list(r.values())); inserted += 1
            except: pass
        cur.execute(f"DROP TABLE {table}")
        cur.execute(f"ALTER TABLE {temp} RENAME TO {table}")
        self.conn.commit()
        ms = (time.perf_counter()-start)*1000
        self.history.append(('full_replace', inserted, ms))
        return inserted
    def append(self, table, records, key_field='order_key'):
        start = time.perf_counter()
        cur = self.conn.cursor()
        ins, skip = 0, 0
        for r in records:
            cur.execute(f"SELECT 1 FROM {table} WHERE {key_field}=?", (r.get(key_field),))
            if cur.fetchone(): skip += 1
            else:
                cols, vals = ', '.join(r.keys()), ', '.join(['?' for _ in r])
                try: cur.execute(f"INSERT INTO {table} ({cols}) VALUES ({vals})", list(r.values())); ins += 1
                except: skip += 1
        self.conn.commit()
        ms = (time.perf_counter()-start)*1000
        self.history.append(('append', ins, ms))
        return ins, skip
    def upsert(self, table, records, key_field):
        start = time.perf_counter()
        cur = self.conn.cursor()
        ins, upd = 0, 0
        for r in records:
            cur.execute(f"SELECT * FROM {table} WHERE {key_field}=?", (r.get(key_field),))
            existing = cur.fetchone()
            if existing:
                ucols = [k for k in r if k != key_field]
                set_c = ', '.join(f"{k}=?" for k in ucols)
                cur.execute(f"UPDATE {table} SET {set_c} WHERE {key_field}=?", [r[k] for k in ucols]+[r[key_field]])
                upd += 1
            else:
                cols, vals = ', '.join(r.keys()), ', '.join(['?' for _ in r])
                cur.execute(f"INSERT INTO {table} ({cols}) VALUES ({vals})", list(r.values()))
                ins += 1
        self.conn.commit()
        ms = (time.perf_counter()-start)*1000
        self.history.append(('upsert', ins+upd, ms))
        return ins, upd
    def scd_type2(self, table, records, key_field='id', tracked=None):
        start = time.perf_counter()
        cur = self.conn.cursor()
        ins, upd = 0, 0
        for r in records:
            kv = r[key_field]
            cur.execute(f"SELECT * FROM {table} WHERE {key_field}=? AND is_current=1", (kv,))
            existing = cur.fetchone()
            if existing:
                changed = any(str(r.get(tf)) != str(existing[tf]) for tf in (tracked or []))
                if changed:
                    cur.execute(f"UPDATE {table} SET is_current=0, valid_to=? WHERE {key_field}=? AND is_current=1",
                        (datetime.now().isoformat(), kv))
                    r['is_current']=1; r['valid_from']=datetime.now().isoformat(); r['valid_to']=None
                    r['scd_version'] = existing['scd_version']+1
                    cols, vals = ', '.join(r.keys()), ', '.join(['?' for _ in r])
                    cur.execute(f"INSERT INTO {table} ({cols}) VALUES ({vals})", list(r.values()))
                    upd += 1
            else:
                r['is_current']=1; r['valid_from']=datetime.now().isoformat(); r['valid_to']=None; r['scd_version']=1
                cols, vals = ', '.join(r.keys()), ', '.join(['?' for _ in r])
                cur.execute(f"INSERT INTO {table} ({cols}) VALUES ({vals})", list(r.values()))
                ins += 1
        self.conn.commit()
        ms = (time.perf_counter()-start)*1000
        self.history.append(('scd_type2', ins+upd, ms))
        return ins, upd

loader = DataLoader(conn)
# 全量覆写
custs = [{'id':i,'name':f'客户_{i}','tier':random.choice(['standard','premium','vip'])} for i in range(1,51)]
r = loader.full_replace('dim_customer', custs)
print(f"全量覆写: {r}条")
# 增量追加
orders = [{'order_key':f'ORD_{i:05d}','customer_id':random.randint(1,50),
    'amount':round(random.uniform(10,5000),2),'status':'pending'} for i in range(1,101)]
ins, skip = loader.append('fact_orders', orders)
print(f"增量追加: 插入{ins}, 跳过{skip}")
# Upsert
upd_orders = orders[:20]
for o in upd_orders: o['status'] = 'delivered'
new_orders = [{'order_key':f'ORD_{i:05d}','customer_id':1,'amount':100,'status':'pending'} for i in range(101,111)]
ins, upd = loader.upsert('fact_orders', upd_orders+new_orders, 'order_key')
print(f"Upsert: 新增{ins}, 更新{upd}")
# SCD Type2
changed = custs[:10]
for c in changed: c['tier'] = 'vip'
ins, upd = loader.scd_type2('dim_customer', changed, 'id', ['tier'])
print(f"SCD Type2: 新增{ins}, 更新{upd}")
# SCD历史
print("SCD历史:")
for row in c.execute("SELECT id,name,tier,scd_version,is_current FROM dim_customer WHERE id<=3 ORDER BY id,scd_version"):
    r = dict(row)
    print(f"  ID={r['id']}, {r['name']}, tier={r['tier']}, v{r['scd_version']}, {'当前' if r['is_current'] else '过期'}")
print("✅ 验证通过 - 数据加载框架运行正常")

📐 批量加载优化

优化点	方法	效果
批量插入	executemany/COPY	10-100x
索引管理	删后重建	2-5x
并行加载	按分区写入	线性扩展

🐍 批量加载性能对比

# 批量加载性能对比
import sqlite3, time, random
conn = sqlite3.connect(':memory:')
c = conn.cursor()
c.execute("CREATE TABLE bench (id INTEGER PRIMARY KEY, value REAL)")
records = [(i, random.uniform(0,100)) for i in range(1,5001)]
c.execute("DELETE FROM bench")
start = time.perf_counter()
for rid, val in records: c.execute("INSERT OR IGNORE INTO bench VALUES (?,?)", (rid, val))
conn.commit()
t1 = (time.perf_counter()-start)*1000
c.execute("DELETE FROM bench")
start = time.perf_counter()
c.executemany("INSERT OR IGNORE INTO bench VALUES (?,?)", records)
conn.commit()
t2 = (time.perf_counter()-start)*1000
print(f"逐条: {t1:.1f}ms, 批量: {t2:.1f}ms ({t1/t2:.1f}x)")
print("✅ 验证通过 - 批量加载对比完成")

💡 实战建议：动手实现本课代码示例，修改参数观察输出变化。生产环境还需考虑错误处理、监控告警和数据安全。

🎯 增量水印

记录最后加载时间
只加载增量
处理乱序数据

🎯 SCD Type3

只保留当前+前一值
对比Type2优劣

🏆 第5课成就解锁

✅ 加载策略：全量/增量/Upsert/SCD
✅ 原子操作：表替换事务保证
✅ 性能优化：批量/索引/并行
✅ SCD维度：历史版本追踪

🎁 下一课预告：数据仓库原理！

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

数据管道设计五原则

幂等性原则：同一输入无论执行多少次，输出结果完全一致。这是重试机制的基础，也是管道可恢复性的保证。
原子性原则：一个批次的数据要么全部成功，要么全部回滚。使用事务或两阶段提交确保原子性。
可观测性原则：每个步骤都有明确的输入输出指标、执行日志和错误追踪。当管道出现问题时，能在5分钟内定位到故障点。
可扩展性原则：管道能水平扩展以应对数据量增长。避免单点瓶颈，使用分区并行处理，设计无状态的计算节点。
自愈性原则：管道能自动处理常见故障，无需人工干预即可恢复。

ETL性能优化清单

优化项	方法	预期效果
I/O优化	批量读写、列式存储、压缩	3-10x
计算优化	向量化执行、谓词下推、列裁剪	2-5x
并行化	分区并行、流水线并行	线性扩展
缓存	维度表缓存、结果缓存	2-10x
增量处理	只处理变更数据	10-100x

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

✅ 管道具有幂等性，可安全重跑
✅ 所有外部依赖有超时和重试机制
✅ 关键步骤有数据量/质量校验
✅ 异常情况有告警通知
✅ 有完整的运行日志和血缘记录
✅ 数据SLA已定义并监控
✅ 回滚方案已测试

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

踩坑与解决方案

在生产环境中，我们经常遇到教科书不会告诉你的问题。以下是常见的实战经验和解决方案：

数据源变更：源系统新增字段或改变格式时，管道不应该崩溃。使用Schema演进策略，新字段设为可选，旧字段保持兼容。
数据倾斜：某些分区数据量远超其他分区，导致处理瓶颈。解决方案：二次分区、Salting技术、自定义分区器。
重复数据：网络重传或重试导致数据重复写入。所有写入操作必须幂等，使用业务主键做去重。
延迟数据：迟到的事件需要特殊处理，不能简单丢弃。使用侧输出收集迟到数据，单独处理。
反压：下游处理速度跟不上上游生产速度。实现背压机制，慢消费者不影响快生产者。

监控与运维指标

指标	告警阈值	监控方式
管道延迟	超过SLA 120%	端到端时间戳追踪
数据量偏差	日均偏差超过30%	与历史同期对比
错误率	超过0.1%	错误计数/总记录数
数据新鲜度	超过SLA 150%	最新数据时间戳
资源利用率	CPU持续>80%	系统监控

⚠️ 常见反模式

❌ 全量表没有分区：查询永远全表扫描
❌ 事实表没有时间维度：无法做时间范围查询
❌ 维表每天全量覆盖：丢失历史变化信息
❌ 管道硬编码配置：环境切换需要改代码
❌ 没有数据质量检查：脏数据直达报表

💡 架构演进建议：不要一开始就追求完美架构。从MVP开始，随业务增长逐步演进。每次演进都要回答三个问题：当前瓶颈是什么？解决它的投入产出比如何？是否为后续扩展留了空间？

第05课：数据加载-批量与增量

📤 数据加载-批量与增量

加载策略对比

🐍 Python实战：核心实战

📐 批量加载优化

🐍 批量加载性能对比

🎯 增量水印

🎯 SCD Type3

🏆 第5课成就解锁

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

数据管道设计五原则

ETL性能优化清单

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

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

踩坑与解决方案

监控与运维指标

📚 延伸阅读与参考

推荐资源