Papermill¶

Apache Airflow 支持与 Papermill 集成。Papermill 是一个用于参数化和执行 Jupyter Notebooks 的工具。也许您有一个财务报告，希望在每个月的第一天或最后一天，或者在年初或年末使用不同的值运行。在您的笔记本中使用参数并使用 PapermillOperator 可以轻松实现这一点。

用法¶

创建笔记本¶

要参数化您的笔记本，请使用标签参数指定一个单元格。Papermill 会查找参数单元格，并将此单元格视为执行时传入参数的默认值。Papermill 将添加一个标记为 injected-parameters 的新单元格，其中包含输入参数，以便覆盖参数中的值。如果没有单元格标记为参数，则注入的单元格将插入到笔记本的顶部。

确保将笔记本保存在 Airflow 可以访问的位置。Papermill 支持 S3、GCS、Azure 和本地。不支持 HDFS。

示例 DAG¶

使用 PapermillOperator 执行 jupyter notebook

tests/system/providers/papermill/example_papermill.py[源代码]

run_this = PapermillOperator(
    task_id="run_example_notebook",
    input_nb="/tmp/hello_world.ipynb",
    output_nb="/tmp/out-{{ execution_date }}.ipynb",
    parameters={"msgs": "Ran from Airflow at {{ execution_date }}!"},
)

用于验证笔记本中消息的示例 DAG

tests/system/providers/papermill/example_papermill_verify.py[源代码]

@task
def check_notebook(inlets, execution_date):
    """
    Verify the message in the notebook
    """
    notebook = sb.read_notebook(inlets[0].url)
    message = notebook.scraps["message"]
    print(f"Message in notebook {message} for {execution_date}")

    if message.data != f"Ran from Airflow at {execution_date}!":
        return False

    return True


with DAG(
    dag_id="example_papermill_operator_verify",
    schedule=SCHEDULE_INTERVAL,
    start_date=START_DATE,
    dagrun_timeout=DAGRUN_TIMEOUT,
    catchup=False,
) as dag:
    run_this = PapermillOperator(
        task_id="run_example_notebook",
        input_nb=os.path.join(os.path.dirname(os.path.realpath(__file__)), "input_notebook.ipynb"),
        output_nb="/tmp/out-{{ execution_date }}.ipynb",
        parameters={"msgs": "Ran from Airflow at {{ execution_date }}!"},
    )

    run_this >> check_notebook(inlets=AUTO, execution_date="{{ execution_date }}")

使用远程 jupyter 内核验证笔记本中消息的示例 DAG

tests/system/providers/papermill/example_papermill_remote_verify.py[源代码]

@task
def check_notebook(output_notebook, execution_date):
    """
    Verify the message in the notebook
    """
    notebook = sb.read_notebook(output_notebook)
    message = notebook.scraps["message"]
    print(f"Message in notebook {message} for {execution_date}")

    if message.data != f"Ran from Airflow at {execution_date}!":
        return False

    return True


with DAG(
    dag_id="example_papermill_operator_remote_verify",
    schedule="@once",
    start_date=START_DATE,
    dagrun_timeout=DAGRUN_TIMEOUT,
    catchup=False,
) as dag:
    run_this = PapermillOperator(
        task_id="run_example_notebook",
        input_nb=os.path.join(os.path.dirname(os.path.realpath(__file__)), "input_notebook.ipynb"),
        output_nb="/tmp/out-{{ execution_date }}.ipynb",
        parameters={"msgs": "Ran from Airflow at {{ execution_date }}!"},
        kernel_conn_id="jupyter_kernel_default",
    )

    run_this >> check_notebook(
        output_notebook="/tmp/out-{{ execution_date }}.ipynb", execution_date="{{ execution_date }}"
    )