python - op_kwargs 中带有 execution_date 的 Airflow Python 脚本

Question

在这个答案的帮助下 https://stackoverflow.com/a/41730510/4200352我正在执行一个 python 文件。

我使用 PythonOperator 并尝试将执行日期作为参数传递给脚本。

我相信我可以通过 kwargs['execution_date'] 以某种方式访问​​它。

以下失败

DAG.py

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

import sys
import os
sys.path.append(os.path.abspath("/home/glsam/OmegaAPI/airflow/scripts/PyPer_ogi_simple"))
from update_benchmarks import *


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 4, 23),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG('run_pyPer', default_args=default_args)

update_BM_G027 = PythonOperator(
    task_id='update_BM_G027',
    python_callable=update_bmk,
    dag=dag,
    op_kwargs={
        'bmk_code': 'G027',
        'is_hedged': False,
        'from_date': kwargs['execution_date'],
    })

也许我需要使用这个答案来获取日期，然后 XCOM 将其用于任务？ https://stackoverflow.com/a/36754930/4200352

Answer 1

这确实有点令人困惑，而且没有很好的记录。

您已经在使用 PythonOperator。

现在只需添加选项

provide_context=True,

并使用指针扩展您的可调用对象，例如

update_bmk(bmk_code, is_hedged, **context)

现在，在您的函数中，您将可以访问有关该任务的所有信息，包括执行日期，如下所示:

task_instance = context['task_instance']
execution_date = context['execution_date']

要查看上下文中项目的完整引用，请参阅 https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html

那些是宏的文档，但您可以使用上下文字典中的项目。