題外話（可以直接跳過進入正題）：

    最近需要使用airflow去排程任務，平常使用airflow都是排程的定時任務，這次不需要定時任務，只需要條件滿足時主動觸發airflow執行任務排程即可，同時將任務必要的外界引數傳給DAG。查了好多資料發現都沒有介紹如何將引數傳給DAG的，於是就去看airflow文件，找到了使用命令列可以將引數傳給DAG。但這種方式不適合我的任務需求，我需要呼叫airflow的API去觸發airflow執行，可是官方文件中這塊介紹的很少很少。也請教了一些人，得到的結果是airflow可能不支援自定義的引數傳入。那怎麼辦呢？難道只能改原始碼？不科學呀，命令列支援，API怎麼會不支援呢？於是本著不行就改原始碼的想法檢視，一路查下去最後發現API和命令列呼叫的是同一個方法（trigger_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": "['
 
[email protected]']",
    "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"}'

引數獲取方式同上

參考：