如何使用 BashOperator 任务的结果作为另一个 Airflow 任务的参数？

Question

我需要将 job_id 参数传递给我的对象 DatabricksRunNowOperator()。 job_id是执行databricks jobs create --json '{myjson}命令的结果。

$ databricks jobs create --json '{myjson}'

{job_id: 12}

import os
import subprocess    
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.databricks_operator import DatabricksRunNowOperator

def pull_function():
    returned_output = subprocess.check_output("echo ti.xcom_pull(key='jobid_CreateCreateRobot')")
    return returned_output


dag_CreateRobot = DAG(dag_id='CreateRobot', 
                      default_args={'owner': 'eric', 
                                  'email': [],
                                  'depends_on_past': False, 
                                  'start_date':'2019-09-16 16:48:28.803023', 
                                  'provide_context': True}, 
                      schedule_interval='@once')

CreateRobot = BashOperator(dag=dag_CreateRobot, 
                                 task_id='CreateRobot', 
                                 bash_command="databricks jobs create --json '{myjson}')")\

RunRobot = DatabricksRunNowOperator(dag=dag_CreateRobot, 
                                    task_id=ti.xcom_pull('RunCreateRobot'), 
                                    job_id=pull_function(), 
                                    databricks_conn_id='myconn', 
                                    json={'token': 'mytoken' })

RunRobot.set_upstream(CreateRobot)

我写这段代码是为了解释我的目标，但它不起作用。如何将 BashOperator 任务的结果用于依赖它的其他任务？

Answer 1

BashOperator中的bash命令需要$ databricks jobs create --json '{myjson}'

即

CreateRobot = BashOperator(dag=dag_CreateRobot, task_id='CreateRobot', bash_command="databricks jobs create --json '{myjson}')", xcom_push=True #Specify this in older airflow versions)

上述运算符在执行时将输出的最后一行推送到 xcom。 (https://airflow.apache.org/_modules/airflow/operators/bash_operator.html)

可以使用 ti.xcom_pull(task_ids='CreateRobot') 访问 xcom 值。

如何使用 BashOperator 任务的结果作为另一个 Airflow 任务的参数？

How to use the result of a BashOperator task as argument of another Airflow task?

airflow

databricks