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 重启后可能从检查点稍早的位置重新处理——这意味着部分数据会重复处理。如果下游不支持幂等就会出现重复数据。常见做法是在写入前用唯一键做 UPSERTINSERT ... 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(parentsTrue, exist_okTrue) # 存储后端 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(fetl: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_idself.task_id, last_offsetoffset, last_keykey, batches_completedself.checkpoint.batches_completed 1 if self.checkpoint else 1, records_processed(self.checkpoint.records_processed records if self.checkpoint else records), last_updatedtime.time(), statusrunning, ) data json.dumps(asdict(self.checkpoint)) if self._use_redis: self.redis.setex( fetl: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(fetl:checkpoint:{self.task_id}) else: cp_file self.checkpoint_dir / f{self.task_id}.json cp_file.unlink(missing_okTrue) 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第 {attempt1} 次重试等待 {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( fETL 完成: 成功 {success_count}, 失败 {failed_count}, f重试 {retry_count} ) except Exception as e: logger.error(fETL 任务异常: {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_okTrue) 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_asciiFalse, defaultstr) 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: fdata_{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_iddaily_sync_0709, batch_size5000) try: etl.run( sourceCursorSource(total50000), transformclean_data, sinkbatch_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%。