如何通过AWS Airflow配置JSON传递参数实现DAG动态触发？

开发者阿泽

2026-7-13

在AWS Airflow中通过触发配置传递参数到DAG脚本

1. 触发时传入配置参数

手动触发DAG时，在AWS Airflow的配置JSON输入框中填入参数，格式如下：

{"message": "你的自定义消息"}

2. 修改DAG代码获取参数

根据你使用的Operator类型，有两种常见实现方式：

方式一：PythonOperator 获取参数

修改你的Hello World DAG代码，通过上下文（context）读取触发配置中的message：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def print_custom_message(**context):
    # 从dag_run的配置中读取message，未传入时用默认值
    custom_message = context['dag_run'].conf.get('message', 'Hello World!')
    print(custom_message)

with DAG(
    'hello_world_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,  # 设置为手动触发
    catchup=False
) as dag:
    print_task = PythonOperator(
        task_id='print_message_task',
        python_callable=print_custom_message,
        provide_context=True  # 必须开启，才能传递上下文到函数
    )

方式二：BashOperator 获取参数

如果用BashOperator执行脚本，可直接通过Airflow模板语法引用参数：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    'hello_world_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    bash_print_task = BashOperator(
        task_id='bash_print_message',
        bash_command='echo "{{ dag_run.conf.get(\'message\', \'Hello World!\') }}"'
    )