第10章 Python ETL实战

第10章：Python ETL实战：使用Python编写完整ETL脚本、调度与日志、错误处理与重试机制

说实话，很多刚入行的朋友问我：「Python写ETL，不就是读个CSV再写进数据库吗？」

嗯，如果只是这样，那确实简单。但真正的生产环境，哪有这么温柔？

数据源可能断连、格式可能突变、网络可能抖动、磁盘可能写满……你写的脚本，得像个老兵一样，能扛得住这些意外。

这一章，我就带你手写一个完整的Python ETL脚本。从数据抽取、转换、加载，到调度、日志、错误处理与重试，一步到位。

10.1 一个完整的ETL脚本长什么样？

我个人习惯把ETL脚本拆成三个核心模块：

抽取（Extract）：从源系统拿数据，可能是数据库、API、文件
转换（Transform）：清洗、过滤、聚合、类型转换
加载（Load）：写入目标系统，比如数据仓库、文件、消息队列

但光有这三个还不够。你还需要：

调度入口：让脚本能被定时或手动触发
日志系统：记录每一步的执行情况，方便排查问题
错误处理与重试：遇到异常别直接崩，先试试能不能自己恢复

下面是一个我常用的脚本骨架：

import logging
import time
from datetime import datetime
from typing import Optional

# ---------- 日志配置 ----------
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'etl_{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# ---------- 抽取 ----------
def extract(source_config: dict) -> Optional[list]:
    """从数据源抽取数据，返回列表或None"""
    try:
        logger.info("开始抽取数据...")
        # 模拟从数据库或API读取
        data = [
            {"id": 1, "name": "张三", "amount": 100.5},
            {"id": 2, "name": "李四", "amount": 200.3},
        ]
        logger.info(f"抽取完成，共 {len(data)} 条记录")
        return data
    except Exception as e:
        logger.error(f"抽取失败: {e}")
        return None

# ---------- 转换 ----------
def transform(data: list) -> list:
    """数据清洗与转换"""
    logger.info("开始转换数据...")
    transformed = []
    for row in data:
        # 类型转换、空值处理、字段映射
        row['amount'] = float(row.get('amount', 0))
        row['etl_time'] = datetime.now().isoformat()
        transformed.append(row)
    logger.info(f"转换完成，共 {len(transformed)} 条")
    return transformed

# ---------- 加载 ----------
def load(data: list, target_config: dict) -> bool:
    """加载到目标系统"""
    try:
        logger.info("开始加载数据...")
        # 模拟写入数据库或文件
        for row in data:
            logger.debug(f"写入: {row}")
        logger.info(f"加载完成，共 {len(data)} 条")
        return True
    except Exception as e:
        logger.error(f"加载失败: {e}")
        return False

# ---------- 主流程 ----------
def run_etl(source_cfg, target_cfg, max_retries=3):
    """带重试机制的ETL主流程"""
    for attempt in range(1, max_retries + 1):
        logger.info(f"=== 第 {attempt} 次尝试 ===")
        data = extract(source_cfg)
        if data is None:
            logger.warning("抽取失败，准备重试...")
            time.sleep(2 ** attempt)  # 指数退避
            continue
        data = transform(data)
        if not load(data, target_cfg):
            logger.warning("加载失败，准备重试...")
            time.sleep(2 ** attempt)
            continue
        logger.info("ETL执行成功！")
        return True
    logger.error("重试次数耗尽，ETL失败")
    return False

if __name__ == "__main__":
    source = {"type": "mysql", "host": "localhost"}
    target = {"type": "postgres", "host": "dw-server"}
    run_etl(source, target)

核心要点：

每个阶段独立成函数，方便单元测试
日志同时输出到文件和控制台，生产环境必备
主流程带重试，避免单次失败就全盘崩溃

10.2 调度：让脚本自己跑起来

脚本写好了，总不能每次都手动执行吧？

我在项目中遇到过，有人用Windows任务计划器跑Python脚本，结果路径写错，日志全丢了……

这里我推荐几种常见的调度方式：

调度方式	适用场景	优点	缺点
Linux Crontab	简单定时任务	轻量、稳定	无依赖管理、无监控
Airflow	复杂DAG依赖	可视化、重试、告警	部署较重
Python schedule库	轻量级脚本内调度	代码内嵌、无需外部依赖	不支持分布式
Kubernetes CronJob	容器化环境	弹性、隔离性好	需要K8s运维能力

如果你只是小团队或个人项目，用 schedule 库就够了。举个例子：

import schedule
import time

def job():
    run_etl(source, target)

# 每天早上8点执行
schedule.every().day.at("08:00").do(job)

while True:
    schedule.run_pending()
    time.sleep(60)

我的建议：生产环境至少用Crontab + 日志监控。如果团队有Airflow，那就用Airflow，它的重试和告警机制非常成熟。

10.3 日志：你的「黑匣子」

日志不是随便print一下就行。我曾经排查一个凌晨3点的ETL失败问题，发现日志里只有一行「Error」，连堆栈都没有……那感觉，就像飞机失事找不到黑匣子。

好的日志系统应该做到：

分级输出：DEBUG、INFO、WARNING、ERROR，别一股脑全打出来
包含上下文：时间、模块、行号、请求ID
自动轮转：日志文件别无限增长，否则磁盘会爆

下面是一个带轮转的日志配置：

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    'etl.log',
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5
)
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)

logger = logging.getLogger('ETL')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

注意：日志级别别设太低。DEBUG日志在线上会刷爆磁盘，我见过有人把DEBUG开了一周，结果100G磁盘全写满了……

10.4 错误处理与重试机制

ETL脚本最怕什么？怕「一次性失败」。比如网络闪断、数据库连接池满、临时锁表……这些错误往往过几秒就能恢复。

所以，重试机制是ETL的标配。

我常用的重试策略有三种：

固定间隔重试：每隔5秒重试一次，最多3次
指数退避：第一次等2秒，第二次等4秒，第三次等8秒……避免雪崩
带抖动的退避：在指数退避基础上加随机值，防止多个任务同时重试

下面是一个带抖动退避的实现：

import random
import time

def retry_with_backoff(func, max_retries=3, base_delay=2):
    for attempt in range(1, max_retries + 1):
        try:
            return func()
        except Exception as e:
            logger.warning(f"第{attempt}次失败: {e}")
            if attempt == max_retries:
                raise
            delay = base_delay ** attempt + random.uniform(0, 1)
            time.sleep(delay)

避坑指南：

不是所有错误都值得重试。比如SQL语法错误、数据格式错误，重试一万次也没用。这种应该直接失败并告警。
重试次数别设太大。我见过有人设了10次重试，结果数据库被重试请求打挂了……
记得记录每次重试的上下文，方便事后分析。

10.5 知识体系总览

下面这张图，是我对本章核心逻辑的总结。你可以把它当作一个「ETL脚本设计检查清单」：

10.6 写在最后

Python写ETL，门槛不高，但要做好却不容易。

你想想看，一个脚本要稳定跑几个月甚至几年，中间可能经历数据源变更、网络波动、磁盘扩容……没有健壮的日志和重试机制，运维起来会非常痛苦。

我个人习惯是：先写日志，再写逻辑；先做错误处理，再做功能。顺序反了，后面补起来很麻烦。

好了，这一章的内容就到这里。代码示例你可以直接拿去用，但记得根据你的实际场景调整配置。

📚 课程目录