Python 数据管线断点续跑：利用本地 SQLite 记录分块状态与失败重试
在处理海量数据批处理（如历史 3 年订单数据回填、数百万篇文档批量 Embedding 向量化、或者跨机房同步千万级日志）时，一个任务往往需要连续运行几个小时甚至整整一夜。
在很多缺乏断点续跑设计的 Python 脚本中：
- 任务以一个巨大的 for chunk in all_chunks: 循环顺序执行；
- 当跑到第 85%（比如已经跑了 3 个小时）时，机房网络偶尔闪断了一下，或者某个外部 API 抛出超时异常，整个脚本直接崩溃退出；
- 开发者第二天早上一看，不仅前面的 3 个小时算力全白费了，重新启动脚本时，由于不知道哪些数据已经写入、哪些没写，只能把数据库清空从第 1 块重新全量跑起！
这种**“一崩全废、无法断点续传”**的脆弱设计，是数据工程中最消耗工程师精力与算力成本的痛点。
今天我们拆解如何利用嵌入式 SQLite 状态机，构建一套轻量、强原子性、零外部依赖的 分块断点续跑与自动恢复框架（Resumable Chunk Pipeline）。
一、断点续跑状态机架构全景
flowchart TD
Init[任务初始化: 将大任务切分为 N 个独立 Chunk] –> RegSQLite[在本地 SQLite 注册所有分块 (状态: PENDING)]
subgraph Run_Loop [断点执行主循环]
RegSQLite –> FetchNext[查询 SQLite 中状态为 PENDING 或 FAILED 的下一分块]
FetchNext –> SetRunning[原子更新状态为 RUNNING, 记录启动时间]
SetRunning –> Process[执行真实的抽取、清洗与写入逻辑]
Process –> CheckSuccess{是否执行成功?}
CheckSuccess — 是 –> SetDone[更新状态为 DONE, 记录处理行数与耗时]
CheckSuccess — 异常失败 –> SetFailed[更新状态为 FAILED, 记录报错日志, 重试计数+1]
end
SetDone –> HasMore{是否还有未完成的分块?}
HasMore — 是 –> FetchNext
HasMore — 否 (全部 DONE) –> Finished[整批任务大功告成!]
Crash[突发断电 / 进程被 Kill] -.-> Resume[重新启动脚本 -> 自动跳过已 DONE 的分块，秒级接续断点!]
二、生产级 SQLite 状态跟踪器代码实现
import sqlite3
import time
import json
import traceback
from typing import List, Dict, Any, Optional
class ChunkStateManager:
def __init__(self, db_path: str = "pipeline_progress.db"):
self.db_path = db_path
self._init_schema()
def _init_schema(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS chunk_status (
chunk_id TEXT PRIMARY KEY,
pipeline_name TEXT NOT NULL,
payload_json TEXT NOT NULL,
status TEXT NOT NULL, — PENDING, RUNNING, DONE, FAILED
retry_count INTEGER DEFAULT 0,
processed_rows INTEGER DEFAULT 0,
error_message TEXT,
started_at TIMESTAMP,
finished_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
def register_chunks(self, pipeline_name: str, chunks: List[Dict[str, Any]]):
"""批量注册任务分块（若已存在则忽略，保障幂等）"""
with sqlite3.connect(self.db_path) as conn:
for c in chunks:
chunk_id = c["chunk_id"]
payload_str = json.dumps(c["payload"], ensure_ascii=False)
conn.execute("""
INSERT OR IGNORE INTO chunk_status (chunk_id, pipeline_name, payload_json, status)
VALUES (?, ?, ?, 'PENDING')
""", (chunk_id, pipeline_name, payload_str))
conn.commit()
def get_next_available_chunk(self, pipeline_name: str, max_retries: int = 3) -> Optional[Dict[str, Any]]:
"""获取下一个待执行的分块（包含 PENDING 或未超重试上限的 FAILED）"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT chunk_id, payload_json, retry_count
FROM chunk_status
WHERE pipeline_name = ?
AND (status = 'PENDING' OR (status = 'FAILED' AND retry_count < ?))
ORDER BY created_at ASC
LIMIT 1
""", (pipeline_name, max_retries))
row = cursor.fetchone()
if not row:
return None
return {
"chunk_id": row[0],
"payload": json.loads(row[1]),
"retry_count": row[2]
}
def mark_running(self, chunk_id: str):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE chunk_status
SET status = 'RUNNING', started_at = CURRENT_TIMESTAMP
WHERE chunk_id = ?
""", (chunk_id,))
conn.commit()
def mark_done(self, chunk_id: str, rows_count: int):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE chunk_status
SET status = 'DONE', processed_rows = ?, finished_at = CURRENT_TIMESTAMP, error_message = NULL
WHERE chunk_id = ?
""", (rows_count, chunk_id))
conn.commit()
def mark_failed(self, chunk_id: str, error_msg: str):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
UPDATE chunk_status
SET status = 'FAILED', retry_count = retry_count + 1, error_message = ?, finished_at = CURRENT_TIMESTAMP
WHERE chunk_id = ?
""", (error_msg, chunk_id))
conn.commit()
三、带断点续跑能力的执行器引擎
def run_resumable_pipeline(pipeline_name: str, all_tasks: List[dict], worker_func):
state_mgr = ChunkStateManager()
# 1. 注册全量分块（已处理过的会自动跳过）
chunks_def = [
{"chunk_id": f"{pipeline_name}_{t['date']}", "payload": t}
for t in all_tasks
]
state_mgr.register_chunks(pipeline_name, chunks_def)
print(f"[*] 任务分块注册完成，准备断点续跑: {pipeline_name}")
# 2. 循环消费待处理分块
while True:
chunk = state_mgr.get_next_available_chunk(pipeline_name, max_retries=3)
if not chunk:
print(f"[✓] 全部分块处理完毕！管线成功结束。")
break
chunk_id = chunk["chunk_id"]
payload = chunk["payload"]
print(f"-> 开始处理分块: {chunk_id} (重试轮次: {chunk['retry_count']})")
state_mgr.mark_running(chunk_id)
try:
# 真实业务计算
affected_rows = worker_func(payload)
state_mgr.mark_done(chunk_id, affected_rows)
print(f"[✓] 分块 {chunk_id} 成功完成，写入 {affected_rows} 行。")
except Exception as e:
err_stack = traceback.format_exc()
print(f"[✗] 分块 {chunk_id} 处理失败: {str(e)}")
state_mgr.mark_failed(chunk_id, err_stack)
time.sleep(2) # 失败微退避
四、生产治理准则
有了这套基于 SQLite 的断点续跑机制，无论面对多么庞大的历史数据回填与网络闪断，你的数据管线都能做到泰然自若、随断随续。
