带配置的气流触发 dag

Question

我尝试使用“trigger w/config”在 dag 中使用配置。

    def execute(**kwargs):
    
    dag_run = kwargs['dag_run']
    start_date = dag_run.conf['start_dt'] if 'start_dt' in dag_run.conf.keys() else kwargs['start_dt']
    end_date = dag_run.conf['end_dt'] if 'end_dt' in dag_run.conf.keys() else kwargs['end_dt']
    
    print(f'start_date = {start_date}, end_date = {end_date}')
dag = DAG(
    "corp_dev_ods_test_dag",
    default_args=default_args,
    description='DAG',
    schedule_interval='10 1 * * *',
    start_date=days_ago(0),
    #params={'dt' : '{{ macros.ds_add(ds, -7) }}'},
    catchup=False,
    tags=['dev']
)
run_submit = PythonVirtualenvOperator(
    task_id='run_submit',
    requirements=dag_requirements,
    python_callable=execute,
    system_site_packages=False,
    dag=dag,
    op_kwargs={'start_dt' : '{{ macros.ds_add(ds, -7) }}', 'end_dt': '{{ macros.ds_add(ds, -7) }}'}
)
run_submit

我收到“KeyError”：kwargs[“dag_run”]。但在 PythonOperator （而不是 PythonVirtualenvOperator）的情况下，它可以工作。

那么，我如何在我的 dag 中使用这些参数呢？

Answer 1

您需要在任务中提供一个空的 params 变量，例如：

from airflow.decorators import dag, task
from datetime import datetime


default_params = {"start_date": "2022-01-01", "end_date": "2022-12-01"}

@dag(
    schedule=None,
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['using_params'],
    params=default_params
)
def mydag():

    @task
    def extract(params={}):
        import helper
        filenames = helper.extract(start=params.get("start_date"))
        return filenames

    extract()

_dag = mydag()

现在在 UI 中，当您使用配置触发 DAG 时，您应该能够查看并更改默认值参数。并且能够在您的 dag 任务中访问它。