Airflow BashOperator collect return code

Question

Bear with me since I've just started using Airflow, and what I'm trying to do is to collect the return code from a BashOperator task and save it to a local variable, and then based on that return code branch out to another task. The issue I have is figuring out how to get the BashOperator to return something. The following is my code segment:

dag = DAG(dag_id='dag_1',
      default_args=default_args,
      schedule_interval='0 2 * * *',
      user_defined_macros=user_def_macros,
      dagrun_timeout=timedelta(minutes=60)
      )
oodas = BashOperator(task_id='oodas', xcom_push=True, bash_command="hive -hiveconf SCHEMA={{ schema }} -hiveconf DAY={{ yesterday_ds }} -f {{ script_path }}", dag=dag)
t2 = BashOperator(task_id='t2', bash_command='echo "{{ ti.xcom_pull("oodas") }}"', dag=dag)
t2.set_upstream(oodas)

I'm trying the xcom_push but honestly have no idea how it works.. Is this the right way to collect the result? In the logs the last line is: Command exited with return code 0.

K-Yo · Accepted Answer · 2018-04-03 16:37:56Z

15

as per the BashOperator doc,

If xcom_push is True, the last line written to stdout will also be pushed to an XCom when the bash command completes

Knowing that, you just need to have your bash script to print the error code last, so append the following to your bash_command :

<your code> ; echo $?

In your case, it is :

oodas = BashOperator(task_id='oodas', xcom_push=True, bash_command="hive -hiveconf SCHEMA={{ schema }} -hiveconf DAY={{ yesterday_ds }} -f {{ script_path }}; echo $?", dag=dag)

answered Apr 3, 2018 at 16:37

K-Yo

1,27010 silver badges15 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Benzi Over a year ago

from version 2.0 and above,(I'm not sure of the specific commit), the last line is automatically pushed and you don't need to write xcom_push=True, see here forum.astronomer.io/t/…

Priyank Mehta · Accepted Answer · 2017-02-17 13:26:15Z

3

Can you post the entire DAG. I think you are having issue in interpreting how Airflow works

From Task1 (if it is a bash operator) you can do :

t1 = BashOperator(task_id='t1', bash_command='echo "{{ ti.xcom_push("t1") }}"', dag=dag)

And in Task2:

t2 = BashOperator(task_id='t2', bash_command='echo "{{ ti.xcom_pull("t1") }}"', dag=dag)

where ti is task_instance variable and {{}} notation is used to access Variables section

answered Feb 17, 2017 at 13:26

Priyank Mehta

2,5432 gold badges23 silver badges34 bronze badges

1 Comment

Ace Haidrey Over a year ago

I just updated more code. So I believe you are right, I'm not using it correctly. Here is my task instance "ti" just "oodas"? I'm not sure what the key is that BashOperator uses for xcom

qalis · Accepted Answer · 2021-11-17 08:49:56Z

-1

For those using Airflow 2+, BashOperator now returns the entire output (source), not just the last line. This way you can not only return error code as K-Yo suggested, but also also perform some additional logging for non-0 return code.

answered Nov 17, 2021 at 8:49

qalis

1,5892 gold badges27 silver badges63 bronze badges

1 Comment

sol25 Over a year ago

I don't think that's correct. Operator uses the subprocess hook and that one return last line only, see: github.com/apache/airflow/blob/…

Collectives™ on Stack Overflow

Airflow BashOperator collect return code

3 Answers 3

1 Comment

1 Comment

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

1 Comment

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related