[Airflow] XCom 을 이용하여 Task 간 데이터 전달

개요

• xcom 은 독립된 task 간 데이터를 공유하는 방법 중 하나
• xcom_push, xcom_pull 을 통해 직렬화 가능한 객체를 저장한다.
  • pickle 로 객채 째로 저장하기 때문에 소규모 데이터에만 사용 권장

task_instance.xcom_push(key="identifier as a string", value=any_serializable_value)

task_instance.xcom_pull(key="identifier as string", task_ids="task-1")

• xcom_pull 호출시 에서 별다른 key 를 명시하지 않으면 return_value 를 기본 key 로 사용
  • do_xcom_push 이 True 일 경우 return_value 가 xcom 에 자동 저장 (기본값 True)

# Pulls the return_value XCOM from "pushing_task"
value = task_instance.xcom_pull(task_ids='pushing_task')

• Xcom 에 불필요한 return_value 가 쌓이는 것을 원하지 않는다면 do_xcom_push 를 False 로 변경할 수 있다.
  • do_xcom_push: operator 의 return value 를 저장하는지 결정하는 옵션 api docs
  • xcom_push, xcom_pull 을 명시적으로 호출하는 것에는 문제 없다.

bash_task = BashOperator(
    task_id='print_date',
    bash_command='echo "This is a test"',
    do_xcom_push=False  # returen_value 저장이 불필요하므로 XCom에 저장하지 않도록 한다. 
)

reference

• Airflow xcoms docs