¿Qué son los Xcoms?

Clase 25 de 29 • Curso de Fundamentos de Apache Airflow

Royer Guerrero Pinilla

student•

✨ La razón por al cual es bash operator comparte su output es porque por defecto tiene el parametro do_xcom_push en True lo mismo podemos hacer con el python operator pero este pusheara lo que coloquemos en el return de la función

Mauro Ezequiel Bravo

student•

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.models.xcom import XCom

default_args = {"depends_on_past": True}

def myfunction(**context):
    print(int(context["ti"].xcom_pull(task_ids='tarea_2')) - 24)

with DAG(dag_id="9-XCom",
    description="Probando los XCom",
    schedule_interval="@daily",
    start_date=datetime(2022, 1, 1),
	default_args=default_args,
    max_active_runs=1
) as dag:

    t1 = BashOperator(task_id="tarea_1",
					  bash_command="sleep 5 && echo $((3 * 8))")

    t2 = BashOperator(task_id="tarea_2",
					  bash_command="sleep 3 && echo {{ ti.xcom_pull(task_ids='tarea_1') }}")

    t3 = PythonOperator(task_id="tarea_3", 
                        python_callable=myfunction)

    t1 >> t2 >> t3

Julio Sarango

student•

Un escenario de uso de Xcoms es retornar por ejm la cantidad de registros cargados a una base de datos que haría una tarea 1 y tomar esta cantidad para el envío de email de notificación desde una tarea 2

Leonardo Martin Mendez

student•

Falta caso de uso de comunicación de tareas ejecutandose en diferentes máquinas, no?

Manuel CABRERA

student•

Hola Profe, agradezco su ayuda, deseo realizar un xcom_pul en un sftpoperator pare no logro hacerlo, agradezco su orientacion porfavor.

download_results = SFTPOperator(
    task_id='download_results',
    ssh_conn_id='my_remote_server',
    remote_filepath="""{{ti.xcom_pull(task_ids='sftp_ostral', key='list_file_xml')}}""",
    local_filepath="""{{ti.xcom_pull(task_ids='sftp_ostral', key='list_file_xml'))}}""",
    operation='get',
)

muchas gracias.

Eric Bellet

teacher•

Hola Andres, qué error estás teniendo?

Federico Martinez

student•

¿Cuándo usar XComs para comunicar?

Utiliza XComs cuando necesites que las tareas compartan información ligera, como IDs, rutas de archivo o resultados pequeños. Son ideales para pasar datos entre operadores que dependen uno del otro en un flujo de trabajo. Evita usarlos para grandes volúmenes de datos, ya que están optimizados para metadatos.

Santiago Ahumada Lozano

student•

XCom Standard vs Custom XCom

Algo importante a mencionar es que las XComs son pensadas para transportar informacion de bajo storage entre tasks. En un backend XCom estandar, la data es serializada y escrita en la base de datos de metadata de airflow. Pero esta base de datos solo guarda strings referenciando bases de datos, por ejemplo una URI.

Se pueden usar XComs para procesar grandes volumenes de datos?

Sí! Estos se denominan Custom XCom Backends. Los cuales permiten guardar datos entre tareas en cualquier ubicacion y formato. Con ello se elimina la limitacion de tamaño de almacenamiento. Para este casos se suelen usar un sistema externo como AWS S3, Google Cloud Storage o Azure blob storage como ubicacion de los datos.

Camilo Corredor

student•

[2025-03-14T18:40:47.370-0500] {xcom.py:690} ERROR - Object of type Resource is not JSON serializable. If you are using pickle instead of JSON for XCom, then you need to enable pickle support for XCom in your *** config or make sure to decorate your object with attr.

Obtuve este error al crear un operador que crea un service google para acceder a un correo electronico. El objeto service que devuelve mi operador no sería JSON serializable. No entiendo por qué? Es decir, no lo puedo compartir con otra tarea?

Cristian Decaroli

student•

Haciendo un proyecto personal, he utilizado xcom para obtener el valor de una task (una función de python) que pushea el valor de un json extraído desde una APi y que luego es consumido por otra task (también función de python) que lo utiliza como argumento. Funcionó perfectamente. Si bien no es lo mejor para enviar 'data', este era un pequeño json y logró enviarse correctamente.

Mario Alexander Vargas Celis

student•

En Apache Airflow, los XComs (abreviatura de "Cross Communications") son una herramienta que permite a las tareas intercambiar mensajes o compartir datos entre sí dentro de un DAG. Esencialmente, los XComs son un mecanismo de paso de mensajes utilizado para transferir pequeños fragmentos de datos entre tareas.

Conceptos Clave de los XComs:

Almacenamiento: Los XComs se almacenan en la base de datos de Airflow.
Tamaño: Están pensados para transmitir datos pequeños, no para manejar grandes volúmenes de información.
Contexto: Usan un contexto de ejecución que incluye atributos como dag_id, task_id, execution_date y key.

¿Cómo funcionan los XComs?

Poner un valor en XCom: Las tareas pueden enviar datos usando el método xcom_push.from airflow.operators.python import PythonOperator from airflow import DAG from datetime import datetime

def push_function(**kwargs): # Empuja un valor al XCom kwargs['ti'].xcom_push(key='my_key', value='Hola, Airflow!')

with DAG( dag_id='xcom_example', start_date=datetime(2024, 1, 1), schedule_interval=None, ) as dag: push_task = PythonOperator( task_id='push_task', python_callable=push_function ) Aquí, my_key es la clave para identificar el valor enviado al XCom.
Obtener un valor de XCom: Una tarea puede recuperar un valor de XCom usando el método xcom_pull.def pull_function(**kwargs): # Recupera el valor desde el XCom value = kwargs['ti'].xcom_pull(key='my_key', task_ids='push_task') print(f'Valor recibido: {value}')

pull_task = PythonOperator( task_id='pull_task', python_callable=pull_function )

push_task >> pull_task En este ejemplo, task_ids='push_task' indica de qué tarea obtener el valor.

Uso con Operadores Bash

Los XComs también pueden ser utilizados con operadores como BashOperator, donde el resultado de un comando Bash se almacena automáticamente en el XCom:

from airflow.operators.bash import BashOperator

bash_task = BashOperator( task_id='bash_task', bash_command='echo "Hola desde Bash"', xcom_push=True # Habilita el envío del resultado al XCom )

python_task = PythonOperator( task_id='read_bash_xcom', python_callable=lambda **kwargs: print(kwargs['ti'].xcom_pull(task_ids='bash_task')) )

bash_task >> python_task

Contextos Comunes de Uso

Transferir datos entre tareas:
- Por ejemplo, una tarea puede descargar datos y otra procesarlos.
Gestionar dependencias dinámicas:
- Usar datos de una tarea anterior para modificar el comportamiento de una tarea posterior.
Combinación con Jinja Templates:t2 = BashOperator( task_id='dynamic_task', bash_command='echo "Valor: {{ ti.xcom_pull(task_ids="push_task", key="my_key") }}"' )
- Los XComs pueden ser utilizados directamente en plantillas Jinja para pasar valores dinámicos a operadores.

Buenas Prácticas

Evita XComs para datos grandes:
- Los XComs no están diseñados para manejar archivos o grandes volúmenes de datos. Usa sistemas externos como S3 o bases de datos para esto.
Uso explícito de claves:
- Asigna claves descriptivas a los XComs para mantener claridad en los datos transferidos.
Limpieza de datos antiguos:
- Los XComs permanecen en la base de datos de Airflow hasta que se eliminan manualmente. Usa estrategias de limpieza para evitar que se acumulen.

from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.models.xcom import XCom

default_args = {"depends_on_past": True}

def myfunction(**context):
    print(int(context["ti"].xcom_pull(task_ids='tarea_2')) - 24)

with DAG(dag_id="9-XCom",
    description="Probando los XCom",
    schedule_interval="@daily",
    start_date=datetime(2022, 1, 1),
	default_args=default_args,
    max_active_runs=1
) as dag:

    t1 = BashOperator(task_id="tarea_1",
					  bash_command="sleep 5 && echo $((3 * 8))")

    t2 = BashOperator(task_id="tarea_2",
					  bash_command="sleep 3 && echo {{ ti.xcom_pull(task_ids='tarea_1') }}")

    t3 = PythonOperator(task_id="tarea_3", 
                        python_callable=myfunction)

    t1 >> t2 >> t3

download_results = SFTPOperator(
    task_id='download_results',
    ssh_conn_id='my_remote_server',
    remote_filepath="""{{ti.xcom_pull(task_ids='sftp_ostral', key='list_file_xml')}}""",
    local_filepath="""{{ti.xcom_pull(task_ids='sftp_ostral', key='list_file_xml'))}}""",
    operation='get',
)

¿Qué son los Xcoms?

Introducción a Apache Airflow

Introducción al curso

¿Para qué sirve airflow?

¿Por qué usar airflow?

Resumen módulo 1

Conceptos básicos

DAG

Tasks y Operators

Scheduler

Instalación y configuración

Instalación de Apache Airflow usando Docker

Posibles configuraciones

Variables y conexiones

Implementando un DAG

Implementando un DAG

Bash Operator

Python Operator

Definiendo dependencias entre tareas

Custom Operator

Orquestar y monitorizar procesos

Orquestando un DAG I

Orquestando un DAG II

Monitoring

Task Actions

Trigger Rules

Sensores

¿Qué son los sensores?

ExternalTaskSensor

FileSensor

Templates con Jinja

¿Qué son los templates con Jinja?

XComs

¿Qué son los Xcoms?

BranchPythonOperator

BranchPythonOperator

Proyecto

Definición del proyecto

Cierre del curso

Cierre del curso

Comparte tu proyecto de Platzi explora el espacio con Airflow y certifícate