4

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.

SpaceyBot
  • 482
  • 1
  • 4
  • 20

3 Answers3

14

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
Ryan Yuan
  • 2,396
  • 2
  • 13
  • 23
  • 1
    Thanks, though it didn't work when i do the same thing inside my bash file instead doing it in bash_command variable itself. Like this: bash_command='/home/me/blah.sh ' where blah.sh has exactly the same piece as what you put inside the ' ' – SpaceyBot Feb 12 '19 at 06:35
  • 2
    It makes sense that it's not working when executing the bash script instead, as the `bash_command` is de-templatized by the Jinja engine **before** being executed. Also, the `xcom_pull` and `xcom_push` are only available in the Airflow context, not in your bash script. Hope that helps – Lucas Jul 18 '19 at 20:06
  • It didn't work in my project. – ramwin Feb 23 '22 at 07:15
  • How would you push the value of a variable in this script instead of passing directly the string? For example, imagine that there was as argument --> `env={"varA":"{{ ds }}" }` and you would like inside the bash command to do a `{{ ti.xcom_push(key="downstreamvar", value=varA }}` When I try to do this I obtain `TypeError: Object of type StrictUndefined is not JSON serializable` – rodvictor May 17 '22 at 10:21
1

@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.

Adrian
  • 171
  • 1
  • 3
  • 3
    I am pretty sure that there is an answer hidden in that wall of text. Please use this info to make it more apparent: https://stackoverflow.com/editing-help – Yunnosch Feb 16 '20 at 22:56
1

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
)
Galuoises
  • 2,630
  • 24
  • 30
  • I'm trying to do something similar but instead of pulling, I want to push the value of the variable `ds` to downstream tasks. I'm trying to do the xcom push with the name of a variable to select the name of the text file to be created and read afterwards but cannot manage to do it. https://stackoverflow.com/questions/72269140/airflow-unable-to-push-templated-env-variable-in-bashoperator-to-xcom – rodvictor May 17 '22 at 10:24