Airflow实战--获取REST参数并通过Xcom传递给Bash算子

Airflow实战–获取REST参数并通过Xcom传递给Bash算子

有时候我们需要通过REST API给调度任务来传递参数。参数可能有多个，也可能只有一个。比如，可能为Spark任务传递一个SQL语句等。

本文分析如何通过REST API为BashOperator的任务来传递参数。

前面已经分析过，如何在PythonOperator任务中来获取REST API的参数，这里分析如何把参数传递给BashOperator任务。

这里我们使用xcom方式来进行任务间的参数传递。也可以使用共享变量的方式来完成该任务，这个后面的文章再介绍。

把参数传递DAG中给不同类型的算子

这里演示：通过Python算子来接收参数，并把参数传递给DAG的Bash算子。

以下代码实现了两个Task，其中一个是PythonOperator，一个是BashOperator。其中的PythonOperator主要是用来接收REST API传递过来的参数值，并把该参值通过xcom保存起来。

BashOperator可以是执行任意任务的shell脚本，它可以通过xcom来获取PythonOperator传递过来的参数。通过这样的方式来实现从REST API给任务传递参数的目的。

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import timedeltadefault_args = {'owner': 'user1','depends_on_past': False,'start_date': airflow.utils.dates.days_ago(1),'email': ['user1@qq.com'],'email_on_failure': False,'email_on_retry': False,'retries': 1,'retry_delay': timedelta(seconds=3),
}dag = DAG('bash_get_param_2',default_args=default_args,description='bash get param job',schedule_interval=timedelta(days=1))# 通过模板的方式来获取xcom传递的参数值。
# 注意：return_value这个变量名是固定的，表示从任务：first_task，过来的变量名
bash_prit_param='echo "I get: {{ ti.xcom_pull(task_ids="first_task", key="return_value") }}"'print_param_task = BashOperator(task_id='print_param_task',do_xcom_push=True,bash_command=bash_prit_param,dag=dag)# 接收REST API的参数，并通过返回参数值来使用xcom保存参数
# 注意：可能有多个参数，可以把这些参数合并到一个变量中，然后返回。若以元组方式返回，则是一个列表。
def receive_param(**context):# 此句用来接收rest接口的参数，参数名为：push_keystock_list = context["dag_run"].conf.get("push_key")print("returned tickers: %s" % str(stock_list))return stock_list# 接收来自rest api的参数和值
first_operator = PythonOperator(task_id='first_task',python_callable=receive_param,dag=dag)print_param_task.set_upstream(first_operator)

代码说明

以上代码需要注意的是Bash算子接收参数的命令，也就是下面的语句：

bash_prit_param='echo "{{ ti.xcom_pull(task_ids="hello_task", key="return_value") }}"'

该语句表示从任务实例（taskinstance）也就是ti中获取返回的参数。这里的ti名称是固定的，表示任务实例；task_ids是返回参数的任务名；key参数的"return_value"也是固定的，表示任务对象的返回参数。

也就是说，通过bash获取参数时，{{ ti.xcom_pull(task_ids="first_task", key="return_value") }}就是bash获取上一个任务的返回值。其中的key的值是固定的。

触发DAG的执行（带参数）

通过以下的请求就可以把参数实时传递给执行的任务。

EXE_DATE=$(TZ=America/Curacao date '+%Y-%m-%dT%H:%M:%SZ')curl -X POST 'http://localhost:20001/api/v1/dags/bash_get_param_2/dagRuns' \-d "{\"execution_date\": \"${EXE_DATE}\", \"conf\": {\"push_key\":\"use db1; show tables;\"}}" \-H 'content-type: application/json' \--user "user1:user1"

命令行说明：以上命令中的conf是传递给远端DAG的参数值。该参数可以是多个，其格式可以不同。我这的参数名是push_key。

小结

本文分析了如何通过REST API给BASH任务传递参数。总结起来，就是：通过python的任务来获取REST API传递过来的参数（可以进行参数检查），并把需要的参数返回（return），这样就是把参数放到了xcom中。然后，其他类型的任务就可以通过模板格式，比如："{{ ti.xcom_pull(task_ids=“hello_task”, key=“return_value”) }}"来获取参数的值。

参数可以是多个，但若python任务返回的参数是多个，可以放到一个变量中，然后返回。