Airflow如何在自定义函数中同时使用kwargs与dag_run上下文参数

Python 不支持在函数定义中同时使用两个 ** 关键字参数接收符，def new_op_fun(**kwargs, **context): 本身就不符合语法规范。实际上你只需要保留一个 ** 接收参数即可：当你给 PythonOperator 设置 provide_context=True 时，所有 Airflow 上下文变量（包括ti、dag_run、execution_date等）都会被打包作为关键字参数传入自定义函数，全部存到你定义的**接收对象里。

另外你现有代码还存在两个可直接修复的错误：

• dag.py 中使用了 BashOperator 但没有导入对应类
• new_op_fun 函数中尝试调用 context["dag_run"]，但你把参数命名为kwargs，代码里没有context这个变量，会报未定义错误

1. dag.py 调整 #

from datetime import datetime
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
# 补全BashOperator导入
from airflow.operators.bash_operator import BashOperator

dag = DAG(
    dag_id='my_dag',
    schedule_interval='@once',
    start_date=datetime(2021, 1, 1)
)
    
previous_op= BashOperator(
    task_id="previous_op",
    bash_command='echo "{{ params.week }}"',
    params = {'week' : '$(date +%V -d \"1 week ago\")',},
    provide_context=True,
    xcom_push=True,
    dag=dag
)

# 两种写法都可以，只要参数名和内部调用一致即可
def new_op_fun(**kwargs):
    # 拉取XCom返回值
    ti = kwargs['ti']
    xcom_value = ti.xcom_pull(task_ids='previous_op')
    # 获取dag_run携带的配置参数
    dag_run_conf = kwargs["dag_run"].conf

    print( xcom_value ) 
    print(dag_run_conf) 
    
    return "hello"

# 如果你习惯用context作为参数名，写法如下，效果完全一致
# def new_op_fun(**context):
#     ti = context['ti']
#     xcom_value = ti.xcom_pull(task_ids='previous_op')
#     dag_run_conf = context["dag_run"].conf
#     print( xcom_value ) 
#     print(dag_run_conf) 
#     return "hello"

new_op = PythonOperator(
    task_id='new_op',
    provide_context=True,
    python_callable=new_op_fun,
    xcom_push=True,
    dag=dag)

previous_op >> new_op 

2. trigger.py 无需调整 #

你现有的 trigger.py 逻辑是正常的，通过modify_dro设置的payload会自动存入被触发DAG的dag_run.conf中，不需要额外修改。

触发DAG后查看new_op任务的日志，就能同时看到previous_op输出的周数XCom值，以及你传入的{"message": "This is my conf message"}配置。

内容的提问来源于stack exchange，提问作者Israel Rodriguez