Передача данных между задачами с помощью XCom

Введение в Apache Airflow на Python

Mike Metzger

Data Engineer

Что такое XCom?

«Cross Communication»
- Позволяет задачам обмениваться данными
Хранится в базе метаданных Airflow
- Передача небольших объёмов данных
- Имена файлов, URI, количество строк

Иллюстрация передачи небольших данных между задачами Airflow через XCom

Что не стоит передавать через XCom

Большие файлы
DataFrames
Полные базы данных
Большие изображения

Иллюстрация типов данных, которые не следует передавать через XCom: большие файлы и DataFrames

Реализация XCom

Существует множество способов использовать XCom
Сосредоточимся на TaskFlow API
Расширение того, что уже сделано с помощью @tasks

Пример XCom

@dag(dag_id='Example_XCom')
def example_xcom():

  @task
  def get_data():
    return data


  @task(multiple_outputs=True)
  def clean_data(sourcedata):
    return clean(sourcedata) # Example, not implemented


  clean_data(get_data())  
example_xcom()

Зависимости XCom

XCom автоматически определяет порядок зависимостей
Пример
```
clean_data(get_data())
```
Концептуально то же самое, что get_data() >> clean_data()

Ещё один пример

 result = clean_data(get_data())
 result >> alert_when_complete()

Просмотр данных XCom

Страница XCom в Airflow со списком сохранённых значений по ключу, DAG и задаче

Давайте потренируемся!

Введение в Apache Airflow на Python