Xcom이란 ?

Cross-Communication이란 뜻으로
Airflow에서 Task 간 데이터 교환(작은 값 전달)을 위한 메커니즘

Task A에서 생성한 데이터를 Task B가 직접 가져와 사용할 수 있다.

저장/조회 API:

xcom_push(key, value) → 값 저장

xcom_pull(task_ids, key) → 값 가져오기

EX) 짝수/홀수 판별

xcom task instance를 통해 값을 주고 받고 최종적으로 홀/짝 판별하는 예제 코드 ( task_ids는 고유학 식별자여야 한다. )

# function definition


# 1. 랜덤 숫자 생성
def generate_number(ti=None):
     num = random.randint(1, 100)
     print(f" random number is {num}")
     ti.xcom_push(key="random_number", value=num) ## ti에 저장 
     # XCom push (key='random_number')

# 2. 제곱 숫자 생성
def square_number(ti=None):
     num = ti.xcom_pull(task_ids="generate_number_task",key="random_number")
     squared = num ** 2  
     print(f" squared number is {squared}")
     ti.xcom_push(key="square_number",value=squared)
     # XCom push (key='square_number')

# 3. 홀/짝판별
def check_even_odd(ti: None):
     num = ti.xcom_pull(task_ids="square_number_task",key="square_number")
     message=""
     if num % 2 == 1 :
          message += f"{num} 은 홀수 입니다."
     else : 
          message += f"{num} 짝수 입니다."
     print(message)

Xcom 데이터 확인

xcom을 사용하면 xcom탭이 하나더 추가 된다

첫번째 task에서 랜덤하게 14가 나왔다.

두번째 task에서는 첫번째 task에서 pull 받은 14*14 = 196이 ti에 저장

세번째 task에서 최종적으로 홀/짝 판별을 통해서 짝수 출력