[Airflow] Bash Operator에서 Xcom 사용

---

Bash Operator에서 Xcom 사용하기

▪ bash operator에서 Xcom을 사용하기 위해서는 env, bash_command 파라미터에서 template을 이용

▪ bash_command 파라미터에 템플릿 문법을 작성해서,  데이터를 Xcom에 올리기

▪ ti 객체에 xcom_push 함수를 이용해서 key, value 형태로 데이터 올리기

▪ bash_command는 출력하는 문장이 return 값으로 간주되어, Xcom에 "return_value" 라는 key 값으로 value가 저장됨

 

▪ env 파라미터에 템플릿 문법을 작성해서, 데이터를 Xcom에서 꺼내기

▪ ti 객체에서 xcom_pull 함수를 이용해서 데이터 꺼내기 

▪ xcom_pull 함수에서 task_ids만 작성을 하면, task에서 return 한 값을 꺼내옴

▪ "do_xcom_push=False"로 지정할 경우, 마지막 return 값이 Xcom에 안 올라감

---

dags_bash_with_xcom.py

▪ BashOperator에서 Xcom을 사용하여 데이터를 올리고 꺼내오는 dag

▪ "bash_push"에서 데이터를 올리고, "bash_pull"에서 데이터를 꺼내와서 출력함

▪ "bash_pull"에서 마지막에 출력되는 값을 Xcom에 올리기 않기 위해, "do_xcom_push=False"로 지정

from airflow import DAG
import pendulum
import datetime
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_with_xcom",
    schedule="10 0 * * *",
    start_date=pendulum.datetime(2023, 12, 10, tz="Asia/Seoul"),
    catchup=False,
) as dag:
    bash_push = BashOperator(
        task_id='bash_push',
        bash_command="echo START && "
                     "echo XCOM_PUSHED "
                     "{{ ti.xcom_push(key='bash_pushed', value = 'first_bash_message') }} &&"
                     "echo COMPLETE"
    )

    bash_pull = BashOperator(
        task_id='bash_pull',
        env={'PUSHED_VALUE':"{{ ti.xcom_pull(key='bash_pushed') }}",
            'RETURN_VALUE':"{{ ti.xcom_pull(task_ids='bash_push') }}"},
        bash_command="echo $PUSHED_VALUE && echo $RETURN_VALUE ",
        do_xcom_push=False
    ) 

    bash_push >> bash_pull

Dag 실행하기

▪ dag을 airflow에 upload하고, webserver 접속해서 dag unpause 

 

▪ bash_push의 실행 결과 log

 

▪ bash_pull의 실행 결과 log

▪ 첫번째 출력은 xcom_push 함수를 사용해서 올렸던 value 값인 first_bash_message

▪ 두번째 출력은 가장 마지막에 return한 값인 COMPLETE

 

 

 

참고:

Airflow 마스터 클래스 강의 - 인프런