如何在Airflow DAG参数的文本字段中传入YAML内容？

直接使用Param的string类型接收完整YAML文本，再在DAG任务内部完成YAML解析即可，具体步骤如下：

1. 定义带字符串类型参数的DAG #

在DAG的params配置里，将接收YAML的参数设为type="string"，支持传入多行YAML文本：

from airflow import DAG
from airflow.models.param import Param
from datetime import datetime
import yaml

with DAG(
    dag_id="yaml_param_dag",
    start_date=datetime(2024, 1, 1),
    params={
        "yaml_content": Param(
            default="""
            key1: value1
            key2:
              - sub_item1
              - sub_item2
            """,
            type="string",
            description="传入完整的YAML文本内容"
        )
    },
    catchup=False
) as dag:
    # 任务定义放在这里

2. 在任务内部解析YAML字符串 #

通过PythonOperator（或其他支持上下文的任务）获取参数中的YAML文本，用pyyaml库解析为Python对象：

from airflow.operators.python import PythonOperator

def process_yaml_data(**context):
    # 从上下文获取传入的YAML文本
    raw_yaml = context["params"]["yaml_content"]
    # 安全解析YAML
    parsed_data = yaml.safe_load(raw_yaml)
    # 后续业务逻辑：比如使用解析后的数据进行处理
    print("解析后的YAML数据:", parsed_data)

# 创建处理任务
yaml_process_task = PythonOperator(
    task_id="parse_and_process_yaml",
    python_callable=process_yaml_data,
    provide_context=True
)

3. 关键注意事项 #

• 确保Airflow运行环境已安装pyyaml，未安装的话执行pip install pyyaml即可
• 传入YAML时要保证格式合法，避免语法错误导致解析失败
• 如果YAML包含特殊字符（如双引号），输入时可使用多行字符串格式（如示例中的三重引号）避免转义问题

内容的提问来源于stack exchange，提问作者Khetag Abramov