airflow - 如何捕获 Airflow 中调用的 DAG 中传递的 --conf 参数

Question

我正在尝试从 REST API 运行 DAG 并向其传递一些参数。 DAG 应该能够捕获参数并使用它。问题是我能够从 REST API 触发 DAG，但 DAG 无法捕获传递的参数。有没有办法实现这个目标？

我从 REST API 触发 DAG，如下所示。它传递 --conf 中的参数

http://abcairflow.com:8090/admin/rest_api/api?api=trigger_dag\&dag_id=trigger_test_dag\&conf=%7B%22key%22%3A%2

如何捕获被调用的DAG中conf值中传递的值。据我所知，conf 应该采用 URL 编码的 JSON 格式数据。

DAG 代码:`

def run_this_func(**kwargs):
print(kwargs)

run_this = PythonOperator(
    task_id='run_this',
    python_callable=run_this_func,
    dag=dag
)`

Answer 1

传递的外部参数是dag_run对象的一部分。可以通过以下方式访问它们:

API Request

import requests

headers = {
    'Cache-Control': 'no-cache',
    'Content-Type': 'application/json',
}

data = '{"conf":"{\\"Key1\\":\\"Value1\\"}"}'

response = requests.post('http://localhost:8080/api/experimental/dags/<dag_id>/dag_runs', headers=headers, data=data)

DAG

 def run_this_func(**context):
    print("Received {} for key=message".format(context["dag_run"].conf))


run_this = PythonOperator(task_id="run_this", python_callable=run_this_func, provide_context=True, dag=dag)