[Airflow] Python 오퍼레이터에서 Xcom 사용

1. Xcom(Cross Communication)

• Airflow DAG 안 Task 간 데이터 공유를 위해 사용되는 기술
  ex) Task1의 수행 중 내용이나 결과를 Task2에서 사용 또는 입력으로 주고 싶은 경우
• 주로 작은 규모의 데이터 공유를 위해 사용
  (Xcom 내용은 메타 DB의 xcom 테이블에 값이 저장됨)
  1GB 이상의 대용량 데이터 공유를 위해서는 외부 솔루션 사용이 필요함 (AWS S3, HDFS 등)
• 크게 두 가지 방법으로 Xcom 사용 가능
  1) **kwargs에 존재하는 ti(task_instance) 객체 활용
     - ti.xcom_push 명시적 사용   (push 방법)
     - ti.xcom_pull 명시적 사용  (pull 방법)
  2) 파이썬 함수의 return 값 활용
     - 함수 return (push 방법)
     - return 값을 input으로 사용 (pull 방법)

 

2. **kwargs에 존재하는 ti(task_instance) 객체 활용 예제 코드

'''
dags_python_with_xcom_eg1.py
'''

from airflow import DAG
import pendulum
import datetime
from airflow.decorators import task

with DAG(
    dag_id="dags_python_with_xcom_eg1",
    schedule="30 6 * * *",
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:

    @task(task_id='python_xcom_push_task1')
    def xcom_push1(**kwargs):
        ti = kwargs['ti']
        ti.xcom_push(key="result1", value="value_1")
        ti.xcom_push(key="result2", value=[1,2,3,])

    @task(task_id='python_xcom_push_task2')
    def xcom_push2(**kwargs):
        ti = kwargs['ti']
        ti.xcom_push(key="result1", value="value_2")
        ti.xcom_push(key="result2", value=[1,2,3,4])


    @task(task_id='python_xcom_pull_task')
    def xcom_pull(**kwargs):
        ti = kwargs['ti']
        value1 = ti.xcom_pull(key='result1') # 가장 마지막에 들어간 result1 값이 pull됨
        value2 = ti.xcom_pull(key='result2', task_ids='python_xcom_push_task1') # python_xcom_push_task1을 task_ids 로 갖는 result2 값이 pull됨
        print(value1)
        print(value2)


    xcom_push1() >> xcom_push2() >> xcom_pull()

 

graph

 

python_xcom_push_task1 XCom

 

python_xcom_push_task2 XCom

 

 

실행결과

 

 

 

 

 

2. 파이썬 함수의 return 값 활용 예제코드

'''
dags_python_with_xcom_eg2.py
'''

from airflow import DAG
import pendulum
import datetime
from airflow.decorators import task

with DAG(
    dag_id="dags_python_with_xcom_eg2",
    schedule="30 6 * * *",
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:

    @task(task_id='python_xcom_push_by_return')
    def xcom_push_result(**kwargs):
        return 'Success'

    @task(task_id='python_xcom_pull_1')
    def xcom_pull_1(**kwargs):
        ti = kwargs['ti']
        value1 = ti.xcom_pull(task_ids='python_xcom_push_by_return')
        print('xcom_pull 메서드로 직접 찾은 리턴 값:' + value1)


    @task(task_id='python_xcom_pull_2')
    def xcom_pull_2(status, **kwargs):
        print('함수 입력값으로 받은 값:' + status)


    python_xcom_push_by_return = xcom_push_result()
    xcom_pull_2(python_xcom_push_by_return)
    python_xcom_push_by_return >> xcom_pull_1()

 

graph

 

 

 

python_xcom_push_by_return Xcom

python_xcom_pull1 실행결과

.

python_xcom_pull2 실행결과