I can't xcom_push an arguments through BashOperator

2
votes

I am new to the Airflow's xcom feature. i tried it out with PythonOperator and it was working fine(i.e., i can push and pull the value out of the context), but when i tried it out on BashOperator, it didn't work. However i can pull only the final stdout statement by adding the xcom_push=True attribute during the task creation. that's one thing. 2) But i also wish to push and pull the values based on their keys (to and from the BashOp) like the way we do it in PythonOp.. It would be really helpful since i need to pass tons of variables from one script to another.

3
airflow

3 Answers

13
votes

Is this what you want?

from datetime import datetime
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator

dag = DAG(
    dag_id="example_bash_operator_1",
    schedule_interval=None,
    start_date=datetime(2018, 12, 31),
)

t1 = BashOperator(
    task_id="t1",
    bash_command='echo "{{ ti.xcom_push(key="k1", value="v1") }}" "{{ti.xcom_push(key="k2", value="v2") }}"',
    dag=dag,
)

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

t1 >> t2
1
votes

@SpaceyBot & Lucas answered your first question. Regarding second question raised

Blockquote

2) But i also wish to push and pull the values based on their keys (to and from the BashOp) like the way we do it in PythonOp.. It would be really helpful since i need to pass tons of variables from one script to another.

Blockquote

  • this is not advisable. All XCom pull/push actions are translated to Insert/Select statements in airflow DB. This will degrade the scheduler performance in time and slow down the whole processing either because of high number of pull(queries) run or the large amounts of rows retrieved which will be retrieved through Full Table scans instead of Index based scans.

So it's better to consider a different mechanism here - storing info in external json/csv/txt files/.. etc.

Bottom line - XCom is designed for transferring small amounts of data only, mostly counters and status variables.

0
votes

In addition to @Ryan Yuan answer you can use the parameter env of the BashOperator to set environmental variables for your bash script/command.

my_task = BashOperator(
    task_id='my_task',
    bash_command='echo $VAR1 $VAR2',
    env={
        "VAR1": '{{ ti.xcom_pull(key="var1")}}',
        "VAR2": '{{ ti.xcom_pull(key="var2")}}'
    },
    dag=dag
)