如何在Python长脚本中实现便捷的检查点（Checkpoints）功能？

嘿，这个场景我太熟了——之前维护过几个小时级的数仓管道，每次出错重跑都要等前面的步骤，简直崩溃！结合你用Jupyter Notebook的交互式场景，给你几个从简单到进阶的实用方案，帮你摆脱重复执行的痛苦：

1. 轻量文件标记法（零依赖，快速上手） #

最直接的思路：给每个步骤设置一个“完成标记文件”，每次运行前先检查这个文件是否存在——存在就跳过该步骤，不存在就执行，执行成功后创建标记，失败则删除标记避免误判。

适合简单的线性管道，不用额外装库，Jupyter里直接就能用：

import os
import time
from pathlib import Path

# 定义检查点目录，统一管理标记文件
CHECKPOINT_DIR = Path("./pipeline_checkpoints")
CHECKPOINT_DIR.mkdir(exist_ok=True)

def step1_extract_data():
    print("执行步骤1：抽取原始数据...")
    # 这里是你的实际数据抽取逻辑，比如读数据库、拉取API
    time.sleep(2)  # 模拟耗时操作
    print("步骤1完成")

def step2_clean_data():
    print("执行步骤2：清洗数据...")
    time.sleep(3)  # 模拟耗时操作
    print("步骤2完成")

# 封装检查点逻辑的装饰器，方便复用
def with_checkpoint(step_name):
    def decorator(func):
        def wrapper():
            checkpoint_file = CHECKPOINT_DIR / f"{step_name}_done.txt"
            if checkpoint_file.exists():
                print(f"✅ 检查点存在，跳过步骤：{step_name}")
                return
            try:
                func()
                # 执行成功，写入检查点
                checkpoint_file.write_text(f"Completed at: {time.ctime()}")
            except Exception as e:
                # 执行失败，删除检查点（如果生成了的话）
                if checkpoint_file.exists():
                    checkpoint_file.unlink()
                raise e  # 抛出异常让你知道哪里错了
        return wrapper
    return decorator

# 给每个步骤加上检查点
step1_extract_data = with_checkpoint("step1_extract")(step1_extract_data)
step2_clean_data = with_checkpoint("step2_clean")(step2_clean_data)

# 运行管道
step1_extract_data()
step2_clean_data()

小贴士：如果需要重新执行某个步骤，直接删除对应的stepX_done.txt文件就行，非常直观。

2. 用缓存库自动保存结果（优雅处理有返回值的步骤） #

如果你的步骤是函数式的（每个步骤输出是下一个步骤的输入），用缓存库会更省心——它会自动把函数的输入输出缓存到磁盘，下次调用相同参数时直接返回缓存结果，不用手动管理标记文件。

推荐两个库：

• joblib：适合numpy/pandas这类大对象的缓存
• diskcache：更灵活，支持更多数据类型

示例：用joblib实现缓存 #

from joblib import Memory
import pandas as pd
import time

# 定义缓存目录
memory = Memory(location="./pipeline_cache", verbose=0)

@memory.cache
def step1_load_data(file_path):
    print("加载数据中...")
    time.sleep(2)
    return pd.read_csv(file_path)

@memory.cache
def step2_preprocess_data(df):
    print("预处理数据中...")
    time.sleep(3)
    df["cleaned_col"] = df["raw_col"].fillna(0)
    return df

# 第一次运行会执行，第二次直接取缓存
raw_df = step1_load_data("raw_data.csv")
clean_df = step2_preprocess_data(raw_df)

优势：不仅跳过步骤，还能直接复用之前的输出结果，省掉重新计算的时间。如果要清空某个函数的缓存，直接调用step1_load_data.clear()就行。

3. 专用工作流工具（适合复杂多分支管道） #

如果你的管道越来越复杂（比如有分支、依赖关系、需要重试/监控），可以用轻量级的工作流工具，它们自带检查点功能，还能在Jupyter里无缝运行。推荐Prefect（比Airflow轻量，适合交互式场景）：

from prefect import flow, task
from prefect.tasks import task_input_hash
from datetime import timedelta
import pandas as pd
import time

# 定义带缓存的任务，设置缓存有效期（比如1天）
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(days=1))
def step1_extract():
    print("抽取数据...")
    time.sleep(2)
    return pd.DataFrame({"raw": [1,2,3,None]})

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(days=1))
def step2_clean(df):
    print("清洗数据...")
    time.sleep(3)
    df["clean"] = df["raw"].fillna(0)
    return df

# 定义工作流
@flow
def data_pipeline():
    raw_data = step1_extract()
    clean_data = step2_clean(raw_data)
    # 后续步骤...
    return clean_data

# 运行工作流，第一次执行，第二次跳过已完成的任务
result = data_pipeline()

亮点：Prefect会自动记录每个任务的状态，失败的任务可以单独重试，还能在UI里查看执行历史——适合团队协作或者长期维护的管道。

Jupyter特有的小技巧 #

在Notebook里，你还可以交互式控制检查点：

• 定义一个全局字典step_status，手动标记步骤是否完成，比如step_status = {"step1": True, "step2": False}，运行前判断状态
• 用%store魔法命令保存步骤状态到Notebook会话外，重启Notebook后还能读取

比如：

# 保存状态
%store step_status
# 读取状态
%store -r step_status

这样在调试的时候，可以灵活选择跳过哪些步骤，不用每次都从头跑。

内容的提问来源于stack exchange，提问作者Peter Dolan