Python ETL 异常恢复：从失败点重试，别每次从头跑
一、跑了 3 小时 40 分钟的任务在第 900 万行失败了，然后从头重跑
ETL 任务最常见的失败模式。
1000 万行数据，前面 900 万行处理正常。
第 9000001 行因为脏数据（空字段、非法格式）抛异常。
整个任务 fail，重试。
又从头开始读那 900 万行。
三个小时后，又卡在第 9000001 行——同一个数据问题没修。
于是工程师手动 Skip 第一条，再跑。
又卡在同批次的另一行脏数据。
这个循环如果发生在凌晨 Cron 任务中，意味着第二天早上数据还没出来。
解决方案不是"手动修数据"。
而是让 ETL 管线支持增量重试和断点续跑。
我在转 Go 之前做了五年 Python 数据处理，这种场景见过不下十次。
每次的心态变化都是一样的：第一次遇到——"数据有问题，修数据"。第二次遇到——"写个 try-except 跳过"。第三次遇到——"加个重试参数"。第 N 次遇到——"这管线得重构"。
但大多数团队在"第三次"就停住了，没有走到"第 N 次"的系统化改造。
原因很简单：重构管线有风险，业务数据要得急，修复当前问题比预防未来问题优先级更高。
这是典型的"温水煮青蛙"——每次手动修一点，直到有一天凌晨三点被 P0 报警叫醒，才发现一整个集群的数据已经停了 8 个小时。
断点续跑的价值不仅仅是节省从头重跑的时间。
它让 ETL 管线从"一次性玩具"变成了"可运维的生产系统"。
有了断点机制，你可以放心地在凌晨自动重试任务，而不是每天早上担心数据有没有跑出来。
二、ETL 断点续跑的核心设计
断点续跑需要两个能力。
一是标记处理进度，二是跳过已处理数据。
flowchart TB
A[启动 ETL 任务] --> B[读取 Checkpoint]
B --> C{有上次 Checkpoint?}
C -->|是| D[从上次位置继续]
C -->|否| E[从头开始]
D --> F[按批次读取数据]
E --> F
F --> G[处理当前批次]
G --> H{处理成功?}
H -->|是| I[更新 Checkpoint]
I --> J[提交批次结果]
H -->|否| K{是可重试错误?}
K -->|是| L[重试当前批次（指数退避）]
K -->|否| M[记录失败批次到死信]
M --> I
J --> N{还有数据?}
N -->|是| F
N -->|否| O[清理 Checkpoint，任务完成]
这个架构有三个核心模块。
第一个是乐观重试（Optimistic Retry）。
对于网络超时、临时的数据库连接断开这类瞬时错误，用指数退避策略重试。
没有退避策略会雪崩——所有失败的任务同时重试，瞬时压力把下游压垮。
第二个是死信队列（Dead Letter Queue）。
重试多次仍然失败的记录，不阻塞管线继续执行，而是写入死信队列。
第二天上班后人工查看死信数据，判断是数据源问题还是代码 bug。
第三个是幂等写入（Idempotent Sink）。
ETL 重启后，可能从检查点稍早的位置重新处理——这意味着部分数据会重复处理。
如果下游不支持幂等，就会出现重复数据。
常见做法是在写入前用唯一键做 UPSERT（INSERT ... ON DUPLICATE KEY UPDATE），而不是简单的 INSERT。
三、断点续跑 ETL 的 Python 实现
下面的代码实现了一个完整的可恢复 ETL 引擎。
支持从文件或 Redis 加载检查点，按批次处理数据，失败记录自动写入死信队列。ResumableETL 的核心逻辑在 run 方法中：先生成 offset，跳过已处理记录，逐批次处理并保存检查点。
"""
resumable_etl.py - 支持断点续跑的 ETL 管线
"""
import json
import logging
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Callable, Iterator, List, Optional, Dict
from contextlib import contextmanager
import redis
logger = logging.getLogger(__name__)
@dataclass
class Checkpoint:
"""ETL 进度检查点"""
task_id: str
last_offset: int # 最后处理的记录偏移量
last_key: Optional[str] # 分片键（如日期）
batches_completed: int # 已完成批次数
records_processed: int # 已处理记录数
last_updated: float # 最后更新时间戳
status: str = "running" # running | completed | failed
class ResumableETL:
"""支持断点续跑的 ETL 引擎"""
def __init__(
self,
task_id: str,
batch_size: int = 10000,
max_retries: int = 3,
checkpoint_dir: str = "./checkpoints",
use_redis: bool = False,
redis_url: str = "redis://localhost:6379/0",
):
self.task_id = task_id
self.batch_size = batch_size
self.max_retries = max_retries
self.checkpoint_dir = Path(checkpoint_dir)
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
# 存储后端
if use_redis:
self.redis = redis.from_url(redis_url)
self._use_redis = True
else:
self.redis = None
self._use_redis = False
self.checkpoint: Optional[Checkpoint] = None
self.dead_letter: List[Dict] = [] # 死信
def load_checkpoint(self) -> Optional[Checkpoint]:
"""加载上次检查点"""
if self._use_redis:
data = self.redis.get(f"etl:checkpoint:{self.task_id}")
if data:
checkpoint = Checkpoint(**json.loads(data))
logger.info(
f"从 Redis 恢复检查点: offset={checkpoint.last_offset}, "
f"已处理 {checkpoint.records_processed} 条"
)
self.checkpoint = checkpoint
return checkpoint
else:
cp_file = self.checkpoint_dir / f"{self.task_id}.json"
if cp_file.exists():
with open(cp_file) as f:
data = json.load(f)
checkpoint = Checkpoint(**data)
logger.info(
f"从文件恢复检查点: offset={checkpoint.last_offset}"
)
self.checkpoint = checkpoint
return checkpoint
return None
def save_checkpoint(self, offset: int, records: int, key: str = None):
"""保存检查点"""
self.checkpoint = Checkpoint(
task_id=self.task_id,
last_offset=offset,
last_key=key,
batches_completed=self.checkpoint.batches_completed + 1
if self.checkpoint else 1,
records_processed=(self.checkpoint.records_processed + records
if self.checkpoint else records),
last_updated=time.time(),
status="running",
)
data = json.dumps(asdict(self.checkpoint))
if self._use_redis:
self.redis.setex(
f"etl:checkpoint:{self.task_id}",
86400 * 7, # 7 天过期
data,
)
else:
cp_file = self.checkpoint_dir / f"{self.task_id}.json"
with open(cp_file, "w") as f:
f.write(data)
logger.debug(
f"检查点已保存: offset={offset}, 记录={records}"
)
def clear_checkpoint(self):
"""清理检查点（任务完成）"""
if self._use_redis:
self.redis.delete(f"etl:checkpoint:{self.task_id}")
else:
cp_file = self.checkpoint_dir / f"{self.task_id}.json"
cp_file.unlink(missing_ok=True)
self.checkpoint = None
def run(
self,
source: Iterator,
transform: Callable,
sink: Callable[[List], None],
key_fn: Callable[[Any], str] = None,
) -> Dict:
"""
执行可恢复 ETL
source: 数据源迭代器（需支持 skip/offset）
transform: 单条记录转换函数
sink: 批量写入函数
key_fn: 记录的分片键函数（用于分片断点）
"""
# 加载检查点
checkpoint = self.load_checkpoint()
start_offset = checkpoint.last_offset if checkpoint else 0
if start_offset > 0:
logger.info(f"从 offset {start_offset} 继续处理")
batch = []
current_offset = start_offset
success_count = 0
failed_count = 0
retry_count = 0
last_error = None
try:
for record in source:
# 跳过已处理记录
if current_offset < start_offset:
current_offset += 1
continue
# 重试策略
for attempt in range(self.max_retries + 1):
try:
result = transform(record)
if result is not None:
batch.append(result)
break # 成功，跳出重试
except Exception as e:
last_error = e
if attempt < self.max_retries:
wait = min(2 ** attempt, 30)
logger.warning(
f"记录 {current_offset} 处理失败，"
f"第 {attempt+1} 次重试，等待 {wait}s: {e}"
)
time.sleep(wait)
retry_count += 1
else:
# 所有重试失败，写入死信
self.dead_letter.append({
"offset": current_offset,
"record": str(record)[:500],
"error": str(e),
})
failed_count += 1
logger.error(
f"记录 {current_offset} 重试全部失败: {e}"
)
current_offset += 1
# 批次满，执行写入
if len(batch) >= self.batch_size:
self._flush_batch(sink, batch, current_offset, record)
success_count += len(batch)
batch = []
# 处理最后一批
if batch:
self._flush_batch(sink, batch, current_offset, record if 'record' in dir() else None)
success_count += len(batch)
# 完成
self.clear_checkpoint()
logger.info(
f"ETL 完成: 成功 {success_count}, 失败 {failed_count}, "
f"重试 {retry_count}"
)
except Exception as e:
logger.error(f"ETL 任务异常: {e}")
# 紧急保存检查点
self.save_checkpoint(current_offset, success_count)
self._save_dead_letter()
raise
return {
"task_id": self.task_id,
"success": success_count,
"failed": failed_count,
"retries": retry_count,
"last_error": str(last_error) if last_error else None,
}
def _flush_batch(
self,
sink: Callable,
batch: List,
offset: int,
last_record: Any,
):
"""安全刷新批次"""
try:
sink(batch)
# 写入成功后才保存检查点
self.save_checkpoint(offset, len(batch))
except Exception as e:
logger.error(f"批次写入失败: {e}")
# 保持检查点不更新，下次从该位置重试
raise
def _save_dead_letter(self):
"""保存死信数据"""
if not self.dead_letter:
return
dl_dir = Path("./dead_letter")
dl_dir.mkdir(exist_ok=True)
dl_file = dl_dir / f"{self.task_id}_{int(time.time())}.json"
with open(dl_file, "w") as f:
json.dump(self.dead_letter, f, ensure_ascii=False, default=str)
logger.warning(f"死信数据已保存: {dl_file} ({len(self.dead_letter)} 条)")
# ---- 使用示例 ----
def demo():
# 模拟数据源：支持从 offset 继续
class CursorSource:
def __init__(self, total: int, start_offset: int = 0):
self.total = total
self.offset = start_offset
def __iter__(self):
for i in range(self.offset, self.total):
yield {"id": i, "value": f"data_{i}"}
# 转换函数（可能失败）
def clean_data(record: dict) -> Optional[dict]:
# 模拟：每逢 10000 的倍数行有一个脏数据
if record["id"] % 10000 == 0 and record["id"] > 0:
if record["id"] == 10000: # 第一个脏数据
raise ValueError(f"数据损坏: id={record['id']}")
return {
"id": record["id"],
"value": record["value"].upper(),
"processed_at": time.strftime("%Y-%m-%d %H:%M:%S"),
}
# 写入函数
def batch_sink(batch: list):
print(f"写入 {len(batch)} 条记录")
etl = ResumableETL(task_id="daily_sync_0709", batch_size=5000)
try:
etl.run(
source=CursorSource(total=50000),
transform=clean_data,
sink=batch_sink,
)
except Exception as e:
print(f"任务中断: {e}")
# 稍后可以重新启动，自动从检查点继续
_flush_batch 方法的"先写入再保存检查点"的顺序是刻意设计的。
如果先保存检查点再写入，写入失败时检查点已经前移——重启后会跳过这批本应重试的数据。
反过来，写入成功后再保存检查点，写入失败时检查点不动，重启后从同一位置重新处理。
代价是可能重复处理，但不会丢数据。
在 ETL 的世界里，"不丢数据"比"不重复处理"重要得多。
四、断点续跑的工程考量
检查点存储需要保证原子性。
写入失败时不更新检查点，下次从旧位置重试。
这样不会丢数据，但可能重复处理。
下游需要支持幂等。
检查点粒度影响恢复效率。
每 1000 条存一次：恢复快但 IO 频繁。
每 50000 条存一次：IO 少但恢复时丢失进度多。
推荐每 5000-10000 条存一次。
不适合的场景：
数据源本身不支持 skip/offset（如实时流）；
数据处理有时效性要求（断点续跑延迟不可接受）；
单条记录处理代价极低的小数据量任务。
在实际生产环境中，还有一个容易被忽略的问题：检查点数据本身的可靠性。
用文件存储检查点时，如果 ETL 任务运行在 Kubernetes 上，Pod 重启后文件可能丢失。
所以对于容器化部署，强烈建议使用 Redis 或数据库存储检查点。
如果必须用文件，把检查点目录挂载到持久卷（Persistent Volume）。
错误分类也是工程实践中的重要细节。
不是所有错误都应该重试。
类型分为三种：
瞬时错误（网络超时、数据库连接断开）→ 重试；
数据错误（脏数据、格式异常）→ 不重试，进死信；
系统错误（内存不足、磁盘满）→ 暂停任务，等人工介入。
五、总结
ETL 断点续跑的核心是检查点机制。
每次批次处理后保存进度到 Redis 或文件。
任务重启时从上次检查点 continue，而非从头开始。
失败记录进入死信队列，不阻塞后续处理。
检查点粒度平衡 IO 频率和恢复效率。
最后给三条可以直接落地的建议：
第一，如果你们的 ETL 任务单次运行超过 30 分钟，断点续跑是刚需，不是可选优化。
第二，不要把检查点和死信写好就完事了，需要配套一个死信查询和重放的运维界面——哪怕只是一个简单的脚本。
第三，检查点的存储介质要和部署架构匹配。单机 cron 用文件，容器用 Redis，分布式用数据库。
做到这三点，凌晨三点的 P0 报警至少能少 80%。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/baronbool/article/details/162717102
