AirFlow 如何传参数给DAG

刺骨的言语ヽ痛彻心扉 2022-04-16 00:48 440阅读 0赞

题外话（可以直接跳过进入正题）：

最近需要使用airflow去调度任务，平常使用airflow都是调度的定时任务，这次不需要定时任务，只需要条件满足时主动触发airflow执行任务调度即可，同时将任务必要的外界参数传给DAG。查了好多资料发现都没有介绍如何将参数传给DAG的，于是就去看airflow文档，找到了使用命令行可以将参数传给DAG。但这种方式不适合我的任务需求，我需要调用airflow的API去触发airflow执行，可是官方文档中这块介绍的很少很少。也请教了一些人，得到的结果是airflow可能不支持自定义的参数传入。那怎么办呢？难道只能改源码？不科学呀，命令行支持，API怎么会不支持呢？于是本着不行就改源码的想法查看，一路查下去最后发现API和命令行调用的是同一个方法（trigger\_dag），既然是同一个方法，肯定支持啊，于是有重新看了一下文档和源码，终于找到了二者之间的联系，哈哈哈，搞定～～

**目录**

1、通过API与AirFlow交互

(1) GET方法获取指定下的指定信息

(2) POST方法运行指定的

2、通过命令行启动DAG时传参

--------------------

## 1、通过API与AirFlow交互 ##

AirFlow提供了两种API交互方式：

### (1) GET方法获取指定<DAG\_ID>下的指定<TASK\_ID>信息 ###

http://<WEBSERVER>:<PORT>/api/experimental/dags/<DAG_ID>/tasks/<TASK_ID>   #returns info for a task (GET).

*  该接口返回的信息主要包括：

{
        "adhoc": "False",
        "depends_on_past": "False",
        "email": "['XXX@XXX.com']",
        "email_on_failure": "True",
        "email_on_retry": "True",
        "end_date": "None",
        "execution_timeout": "None",
        "max_retry_delay": "None",
        "on_failure_callback": "None",
        "on_retry_callback": "None",
        "on_success_callback": "None",
        "op_args": "[]",
        "op_kwargs": "{}",
        "owner": "XXX",
        "params": "{}",
        "pool": "None",
        "priority_weight": "1",
        "provide_context": "True",
        "python_callable": "<function get_params at 0x7f2e156691b8>",
        "queue": "3708",
        "resources": "{'disk': {'_qty': 512, '_units_str': 'MB', '_name': 'Disk'}, 'gpus': {'_qty': 0, '_units_str': 'gpu(s)', '_name': 'GPU'}, 'ram': {'_qty': 512, '_units_str': 'MB', '_name': 'RAM'}, 'cpus': {'_qty': 1, '_units_str': 'core(s)', '_name': 'CPU'}}",
        "retries": "1",
        "retry_delay": "0:05:00",
        "retry_exponential_backoff": "False",
        "run_as_user": "XXX",
        "sla": "None",
        "start_date": "2018-07-09 00:00:00",
        "task_id": "get_params",
        "templates_dict": "None",
        "trigger_rule": "all_success",
        "wait_for_downstream": "False"
    }

### (2) POST方法运行指定的<DAG\_ID> ###

http://<WEBSERVER>:<PORT>/api/experimental/dags/<DAG_ID>/dag_runs   #creates a dag_run for a given dag id (POST).

*  传入自定义参数给DAG

header: Content-Type:application/json

{
        "conf": 自定义参数（json格式字符串）
    }

注：可传入的参数包括run\_id、conf、execution\_date

例如：

{
        "conf": "{\"id\":1, \"start_time\":\"2018-01-01 00:00:00\", \"end_time\":\"2018-10-31 23:59:59\"}"
    }

*  通过以上方式传入的参数会存放在 ‘dag\_run’的conf属性中，因此在DAG中获取参数方式：

def get_params(**kwargs):
        id=kwargs.get('dag_run').conf.get('id')
        start_time=kwargs.get('dag_run').conf.get('start_time')
        end_time=kwargs.get('dag_run').conf.get('end_time')

注：同一个DAG中的每个task只要传入\*\*kwargs，都可以通过该方式获取

## 2、通过命令行启动DAG时传参 ##

$ airflow trigger_dag <dag_id> -c <json格式的字符串参数>

例如

$ airflow trigger_dag <dag_id> -c '{"id":1, "start_time":"2018-01-01 00:00:00", "end_time":"2018-10-31 23:59:59"}'

参数获取方式同上

参考：

\[1\] [http://airflow.incubator.apache.org/api.html][http_airflow.incubator.apache.org_api.html]

\[2\] [http://airflow.incubator.apache.org/cli.html\#][http_airflow.incubator.apache.org_cli.html]

\[3\] [https://github.com/apache/incubator-airflow][https_github.com_apache_incubator-airflow]

[http_airflow.incubator.apache.org_api.html]: http://airflow.incubator.apache.org/api.html
[http_airflow.incubator.apache.org_cli.html]: http://airflow.incubator.apache.org/cli.html#
[https_github.com_apache_incubator-airflow]: https://github.com/apache/incubator-airflow