#apacheairflow

Если Postgres - это надежный банковский сейф, где каждая транзакция на вес золота, то ClickHouse - это промышленная мясорубка. Ему все равно, уникальны ли ваши записи (по умолчанию), он не поддерживает классические транзакции, но зато он умеет делать SELECT count(*) FROM hits по миллиарду строк за доли секунды. Для инженера Airflow работа с ClickHouse кардинально отличается от работы с обычными реляционными базами. Главное правило ClickHouse: Никогда не вставляйте данные по одной строке. Если вы напишете цикл в Python, который делает INSERT INTO table VALUES (...) миллион раз, вы положите кластер. ClickHouse любит, когда в него вставляют данные большими кусками (батчами) по 10–100 тысяч строк за раз. И Airflow должен уметь это организовать.  

Шаг 1. Добавляем ClickHouse в инфраструктуру ETL pipeline Airflow

  Расширим наш docker-compose.yaml. ClickHouse очень экономен к ресурсам, поэтому для тестов нам хватит минимальной конфигурации. Добавьте этот сервис: clickhouse: image: clickhouse/clickhouse-server:latest ports: - "8123:8123" # HTTP порт (для веб-клиентов и некоторых драйверов) - "9000:9000" # Нативный TCP порт (самый быстрый, для Python-драйвера) ulimits: nofile: soft: 262144 hard: 262144 healthcheck: test: interval: 30s timeout: 10s retries: 3 Не забудьте docker-compose up -d. Проверить работу можно, открыв localhost:8123 в браузере (должен ответить "Ok"). Также нам понадобится провайдер для Airflow. Добавьте в ваш Dockerfile: RUN pip install apache-airflow-providers-clickhouse clickhouse-driver И пересоберите образ.  

Шаг 2. Настройка Connection: HTTP vs Native

  В Airflow есть путаница с типами подключений к ClickHouse. HTTP (порт 8123): Проще, работает через requests. Надежно, но чуть медленнее на огромных объемах. Native (порт 9000): Работает через бинарный TCP-протокол. Это выбор чемпионов. Библиотека clickhouse-driver использует именно его. Настроим соединение clickhouse_native. Admin -> Connections -> Add Conn Id my_clickhouse Conn Type ClickHouse (если провайдер установлен корректно) Host clickhouse (имя сервиса Docker) Login/Password default / (пусто), если не меняли настройки Port 9000 (для нативного протокола)  

Шаг 3. Практика: Загрузка данных из S3 в ClickHouse

  У нас есть два пути загрузки данных, и выбор зависит от объема. "Ленивый" (через движок S3) ClickHouse настолько крут, что умеет сам ходить в S3 и забирать данные, вообще не нагружая Airflow. Airflow просто посылает команду: "Эй, ClickHouse, вот бакет, забери файлы". Это лучший способ для больших данных (ГБ и ТБ). "Классический ETL" (Python Driver) Airflow читает файл, преобразует его (например, меняет формат дат) и вставляет в ClickHouse. Этот способ мы разберем подробно, так как он учит работать с батчами и хуками. Напишем DAG, который берет CSV из S3 (результат прошлых статей) и вставляет его в таблицу user_stats.   Подготовка таблицы (DDL)   Сначала создадим таблицу. Обратите внимание на движок MergeTree - это стандарт для аналитики.   CREATE TABLE IF NOT EXISTS user_stats ( date Date, name String, count UInt32 ) ENGINE = MergeTree() ORDER BY date; Код Airflow DAG: s3_to_clickhouse.py   from airflow import DAG from airflow.operators.python import PythonOperator from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook # Или стандартный from airflow.providers.amazon.aws.hooks.s3 import S3Hook from datetime import datetime import io import csv def load_data_to_clickhouse(**context): # 1. Читаем данные из S3 s3_hook = S3Hook(aws_conn_id="minio_s3") bucket = "airflow-bucket" key = "users_export_2023-01-01.csv" # В реальности используйте шаблоны {{ ds }} # Скачиваем файл в память (для больших файлов лучше стримить или качать на диск!) obj = s3_hook.get_key(key, bucket) file_content = obj.get().read().decode('utf-8') # Парсим CSV в список кортежей # ClickHouse драйвер ждет список: data = reader = csv.DictReader(io.StringIO(file_content)) for row in reader: data.append(( row, row, int(row.get('count', 1)) # Защита от пустых значений )) print(f"Подготовлено {len(data)} строк для вставки.") # 2. Вставляем в ClickHouse # Используем execute с параметром params для bulk-вставки ch_hook = ClickHouseHook(clickhouse_conn_id="my_clickhouse") sql = "INSERT INTO user_stats (date, name, count) VALUES" # Магия clickhouse-driver: мы передаем список данных вторым аргументом. # Драйвер сам разобьет это на блоки и отправит бинарным потоком. # Это В РАЗЫ быстрее, чем циклы INSERT. ch_hook.execute(sql, data) print("Вставка завершена.") with DAG( dag_id="s3_to_clickhouse_loader", start_date=datetime(2023, 1, 1), schedule=None, catchup=False ) as dag: # 0. Создаем таблицу (лучше вынести в отдельный скрипт миграций, но для теста сойдет) create_table = PythonOperator( task_id="init_table", python_callable=lambda: ClickHouseHook(clickhouse_conn_id="my_clickhouse").execute( "CREATE TABLE IF NOT EXISTS user_stats (date Date, name String, count UInt32) ENGINE = MergeTree() ORDER BY date" ) ) # 1. Грузим данные load_task = PythonOperator( task_id="load_from_s3", python_callable=load_data_to_clickhouse ) create_table >> load_task      

Тонкости и подводные камни

Работа с ClickHouse в Airflow полна нюансов, о которых не пишут в Quickstart-гайдах. Проблема идемпотентности (Дубликаты) ClickHouse не проверяет уникальность (Primary Key) при вставке в обычный MergeTree. Если вы запустите DAG два раза, у вас будет двойной объем данных. Решение для новичков: Перед вставкой делать ALTER TABLE ... DELETE WHERE date = '{{ ds }}'. Но в ClickHouse операции удаления (Mutation) - тяжелые и асинхронные. Решение для профи: Использовать движок ReplacingMergeTree (он схлопывает дубликаты в фоне) или вставлять данные во временную таблицу, а потом делать EXCHANGE PARTITION (атомарная замена куска данных). Типизация Postgres простит вам, если вы передадите число как строку "123". ClickHouse при вставке через нативный протокол строг. Если колонка UInt32, а вы суете str, драйвер упадет. Всегда явно приводите типы в Python (как мы сделали int(row)). Таймауты ClickHouse быстрый, но если вы попытаетесь вставить 10 ГБ одним запросом, соединение может разорваться. Совет: Разбивайте данные на чанки (chunks) по 10–50 тысяч строк внутри Python-кода и делайте ch_hook.execute в цикле.   Альтернатива: ClickHouseOperator   В провайдере есть готовый ClickHouseOperator. Он удобен для простых SQL-команд (оптимизация, удаление, создание таблиц).   from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator optimize_table = ClickHouseOperator( task_id="optimize_user_stats", clickhouse_conn_id="my_clickhouse", sql="OPTIMIZE TABLE user_stats FINAL" ) Используйте его для сервисных задач, а загрузку данных делайте через Python/Hooks, так как вам нужен контроль над форматом данных. Исправьте финальные варианты кода Dags  и  конфигурационных файлов и при необходимости сравните с нашими на GitHub где лежит код к Уроку 9.  

Помощь Cursor: Генерация SQL и кода вставки

ClickHouse SQL (диалект) местами специфичен. Cursor поможет не лезть в документацию за синтаксисом движков. Промпт 1 (DDL):   "Напиши SQL для создания таблицы ClickHouse events, которая хранит логи веб-сайта (timestamp, user_id, url). Используй движок MergeTree, партиционирование по месяцам и TTL (время жизни), чтобы удалять данные старше года."     Промпт 2 (Оптимизация вставки): "Посмотри на этот Python-код вставки в ClickHouse. Перепиши его так, чтобы использовать генератор (generator) и вставлять данные батчами по 20 000 строк, чтобы не перегружать оперативную память."   Итог: Мы построили полный цикл: Данные -> Postgres -> S3 -> Обработка -> ClickHouse. Теперь в нашей базе лежат "золотые" данные, готовые к построению графиков в Grafana или Superset. Но есть одна проблема, с которой вы столкнетесь, когда дагов станет 50 штук. Как не писать один и тот же код 50 раз? Как создавать DAG-и динамически, на основе конфигурационных файлов, а не копипасты? В финальной статье мы поговорим про Best Practices, динамическую генерацию DAG-ов и организацию "чистого кода" в Airflow. Готовы к рефакторингу и высшему пилотажу?    

Использованные референсы и материалы

До этого момента все наши DAG-и жили по расписанию. schedule_interval='@daily' - это классика. Но современный бизнес не хочет ждать "утреннего отчета". Если данные прилетели в 14:00, отчет должен быть готов в 14:10, а не на следующее утро. Здесь мы сталкиваемся с фундаментальным конфликтом: Airflow - это Batch-инструмент (запускает задачи пачками). Kafka - это Streaming-инструмент (поток событий). Многие новички пытаются обрабатывать сообщения из Kafka прямо внутри Airflow (например, в цикле PythonOperator). Это архитектурная ошибка. Airflow не предназначен для вычитывания миллионов сообщений в секунду. Правильный паттерн использования Airflow с Kafka - это Event-Based Triggering (Запуск по событию). Сценарий: В Kafka падает сообщение: {"status": "batch_ready", "s3_path": "s3://bucket/data.csv"}. Airflow "слушает" топик. Как только видит это спец-сообщение - запускает тяжелый DAG на Spark или Dask для обработки указанного файла.  

Как Airflow может "слушать" сообщения в топике Kafka

  Есть три способа заставить Airflow ждать события, и два из них - плохие. Сенсор в режиме mode="poke" (Плохо) Воркер занимает слот, подключается к Kafka и в бесконечном цикле спрашивает: "Есть сообщение? Нет. А сейчас? Нет.". Вы сжигаете ресурсы воркера. Если у вас 10 сенсоров и 10 слотов на воркере, весь кластер встанет в ожидании, и полезная работа не будет делаться. Сенсор в режиме mode="reschedule" (Лучше) Воркер проверяет Kafka. Если пусто - задача "умирает" (освобождает слот) и планирует следующую проверку через 5 минут. Задержка реакции. Сообщение пришло в 12:01, а проверка будет только в 12:05. Deferrable Operators / Triggers (Идеально) Это технология Airflow 2.2+. Задача освобождает воркер полностью и передает обязанность ожидания специальному легкому сервису - Triggerer. Один процесс Triggerer может асинхронно (через AsyncIO) ждать тысячи событий одновременно, почти не потребляя память. Нужно запустить еще один контейнер triggerer в инфраструктуре. В этой статье мы реализуем Сенсор, так как это база, но будем держать в уме, что в HighLoad-системах его нужно переписывать на Триггер.      

Шаг 1. Добавляем Kafka в песочницу AirFlow

  Нам нужен брокер сообщений. Добавим Kafka без Zookeeper (он теперь не нужен для управления Кафкой) в docker-compose.yaml. kafka: profiles: - broker image: apache/kafka:3.9.0 container_name: kafka ports: - "9092:9092" environment: CLUSTER_ID: "MkU3OEVBNTcwNTJENDM2Qk" KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://kafka:9092,CONTROLLER://kafka:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092 KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 Также нам нужно научить Airflow работать с Kafka. Добавьте библиотеку в Dockerfile: RUN pip install confluent-kafka (или используйте готовый провайдер apache-airflow-providers-apache-kafka, но для обучения мы напишем логику руками, чтобы понять суть).

Не забудьте пересобрать образы: docker-compose up -d --build. В случае если провайдер Kafka не появляется, потребуется жесткая пересборка имиджа airflow-hdfs:2.8.1 docker build --no-cache -t airflow-hdfs:2.8.1 . docker compose up -d --force-recreate Теперь главное проверим что все необходимое работает и в сборе. Пробуем зайти в docker Kafka

#--- подключаемся внутрь консоли docker container c Kafka и создаем топик для нашего DAG "etl_start" docker exec -it kafka bash bin/kafka-topics.sh --bootstrap-server kafka:9092 --create --topic etl_start #--- подключаемся продюсером из консоли кафка и кидаем сообщение для запуска процессинга {"status": "ready", "date": "2026-02-03"} bin/kafka-console-producer.sh --bootstrap-server kafka:9092 --topic etl_start    

Шаг 2. Как Airflow "потребляет" события из топика Apache Kafka

  Мы не будем читать все данные. Мы будем ждать сигнал. Для этого напишем кастомный сенсор. Почему кастомный? Потому что стандартные сенсоры часто либо слишком сложны в настройке, либо не поддерживают специфическую логику (например, "прочитать JSON и проверить поле status"). Создайте папку plugins/sensors (если нет) и файл kafka_custom_sensor.py: import json import logging from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.apache.kafka.sensors.produce_consume import AwaitMessageTriggerFunctionSensor from datetime import datetime # Функция-обработчик. Airflow будет передавать в нее каждое новое сообщение из топика. # Важно: она должна лежать на уровне модуля, чтобы Airflow мог ее сериализовать. def check_event(message): try: # message - это объект confluent_kafka.Message val = message.value().decode('utf-8') data = json.loads(val) if data.get("status") == "ready": logging.info("Отмашка получена, запускаем обработку.") return True # Возвращаем True, сенсор успешно завершает работу except Exception as e: logging.warning(f"Пришло сообщение в другом формате, игнорируем: {e}") return False # Возвращаем False, сенсор продолжает слушать эфир def process_data(): logging.info("Имитация тяжелой обработки данных...") with DAG( dag_id="08.kafka_event_driven", start_date=datetime(2023, 1, 1), schedule=None, catchup=False ) as dag: wait_for_event = AwaitMessageTriggerFunctionSensor( task_id="wait_for_kafka_msg", kafka_config_id="kafka_default", # Ссылка на подключение в Airflow topics=, # Здесь указываем путь к нашей функции проверки. # Если файл называется 08_kafka_event_driven.py, то путь такой: apply_function="08_kafka_event_driven.check_event", poll_interval=10, poll_timeout=3600, ) run_job = PythonOperator( task_id="process_data", python_callable=process_data ) wait_for_event >> run_job  

Шаг 3. Реализация DAG Airflow - Event-Driven Pipeline из Kafka

  Допустим внешняя система загрузила файл и кинула событие в топик etl_start. Мы ждем это событие и запускаем обработку. Вот финальный код для файла 08_kafka_event_driven.py, который решает эту проблему. import json import logging from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.apache.kafka.sensors.kafka import AwaitMessageSensor from datetime import datetime # Функция для фонового процесса (триггера) def check_event(message): try: val = message.value().decode('utf-8') data = json.loads(val) if data.get("status") == "ready": logging.info("Сигнал получен, завершаем ожидание") return data except Exception as e: logging.warning(f"Неизвестный формат сообщения, пропускаем: {e}") # Явный возврат пустоты заставляет сенсор ждать следующее сообщение return None def process_data(): logging.info("Имитация тяжелой выгрузки данных") with DAG( dag_id="08.kafka_event_driven", start_date=datetime(2023, 1, 1), schedule=None, catchup=False, tags= ) as dag: wait_for_event = AwaitMessageSensor( task_id="wait_for_kafka_msg", kafka_config_id="kafka_default", topics=, apply_function="08_kafka_event_driven.check_event", poll_interval=10, poll_timeout=3600, ) run_job = PythonOperator( task_id="process_data", python_callable=process_data ) wait_for_event >> run_job Как это работает в динамике Вы запускаете DAG. Задача wait_for_kafka_msg переходит в статус UP_FOR_RESCHEDULE. Она не занимает слот воркера.

Каждые 30 секунд Airflow "просыпается", запускает мини-задачу проверки Kafka и засыпает снова. Вы (через консоль или скрипт) отправляете сообщение в Kafka: {"status": "ready", "file_id": "file_123"}. При следующей проверке Сенсор ловит JSON, видит status: ready, возвращает True. DAG переходит к задаче process_data, которая печатает "запускает task execution".

    Проблема "Consumer Group" в контексте Airflow   Это самый тонкий момент, о который спотыкаются все инженеры. Kafka помнит, что "группа пользователей" прочитала сообщение, сдвигая offset (указатель). Если ваш DAG упал и перезапустился с тем же group.id, он не увидит старое сообщение, потому что Kafka считает, что "вы его уже читали". Для событий: Часто используют auto.offset.reset': 'latest', чтобы реагировать только на то, что происходит сейчас. Для гарантии доставки: Если важно не пропустить ни одного сигнала (даже если Airflow лежал), нужно управлять оффсетами вручную или использовать уникальные group.id для каждого запуска (что засоряет Kafka).   Deferrable Operators - будущее уже здесь  

Код классического сенсора хорош, но он все равно дергает планировщик каждые 30 секунд. Если таких сенсоров тысячи, планировщик задохнется. В современном Airflow с провайдером apache-airflow-providers-apache-kafka архитектура работает иначе. Разница подходов:

-

Sensor - Воркер спрашивает про наличие данных. Если их нет, он уходит в сон, потом просыпается и спрашивает снова.

-

Trigger - Воркер просит триггер разбудить его при появлении нужного сообщения и полностью освобождает ресурсы. Отдельный сервис держит открытым сокет к Kafka. Как только байты прилетели, триггер дает сигнал Airflow, и воркер возвращается к работе.

-

На практике это выглядит так: задача переходит в режим ожидания и окрашивается в специфический бледно-зеленый цвет статуса deferred. В этот момент она не потребляет вычислительные ресурсы кластера. Для новичка настройка Triggerer может быть сложной, но знать о ней обязательно. При переходе на этот механизм критически важно правильно выбрать класс оператора. Библиотека предлагает два похожих решения, работу которых легко спутать при чтении логов.

Класс AwaitMessageTriggerFunctionSensor спроектирован как бесконечный слушатель. Его цель заключается в непрерывном мониторинге топика. При появлении сообщения он вызывает функцию проверки, отрабатывает логику и моментально возвращается на пост дожидаться следующего сигнала. Такой инструмент никогда не переходит в статус успеха по своей воле.

В логах это выглядит как постоянный уход в спячку. Задача ловит сообщение, рапортует об успехе, но затем выдает код возврата 100 (task deferral) и снова ставится на паузу.

Для классического конвейера данных требуется другой подход. Нам нужно дождаться одного конкретного события, отметить шаг выполненным и передать эстафету следующей задаче графа. С этой задачей отлично справляется AwaitMessageSensor. Как только логика нашей проверки явно возвращает любые данные вместо пустоты, он понимает, что цель достигнута, прекращает прослушивание и закрывает задачу с зеленым статусом успеха.

Лог наглядно подтверждает разницу. Как только нужное сообщение найдено, процесс выходит из режима deferral, задача помечается успешной, и конвейер идет дальше без зацикливания.

Исправьте финальные варианты кода Dags, сенсора  и  конфигурационных файлов и при необходимости сравните с нашими на GitHub где лежит код к Уроку 8.   Troubleshooting: Когда тишина в эфире   Ошибка 1: cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"} Причина: Airflow не может достучаться до Kafka. Лечение: Проверьте bootstrap.servers. Если вы внутри Docker, это должно быть имя сервиса (kafka:9092). Если снаружи - localhost:9092 (при условии правильной настройки ADVERTISED_LISTENERS). Ошибка 2: Сенсор вечно висит, хотя сообщение отправлено Причина: Проблема с сериализацией или условием. Вы отправили строку "ready", а код ждет JSON {"status": "ready"}. Лечение: Добавьте логирование self.log.info(f"Raw msg: {msg.value()}") до этапа парсинга JSON.  

Помощь Cursor: Генерация продюсера Kafka для DAG Apache AirFlow 

  Вам нужно как-то тестировать сенсор, отправляя сообщения. Промпт для Cursor: "Напиши простой Python-скрипт используя библиотеку confluent-kafka, который отправляет JSON сообщение {'status': 'ready', 'date': '2023-10-20'} в топик 'etl_start'. Настрой подключение на localhost:9092." Итог: Мы сделали Airflow реактивным. Теперь он не просто молотит по расписанию, а ждет команды от внешнего мира через шину событий. Это и есть современная "Event-Driven" архитектура. В следующей статье мы отойдем от технологий передачи данных и займемся аналитикой. Мы подключим Airflow к ClickHouse - самой быстрой базе данных для OLAP-запросов. Мы разберем, как вставлять миллионы строк за секунды, используя нативный протокол, и почему PostgresOperator здесь не подойдет. Готовы к скорости ClickHouse?    

Использованные референсы и материалы

В прошлых статьях мы выяснили: если задача тяжелая и требует Java (Spark), мы используем SparkSubmitOperator. Но что делать, если у вас "тяжелый" Python? Типичная ситуация когда вы написали отличный код на Pandas внутри PythonOperator. На тестовом файле в 100 Мб все летало. В продакшене пришел файл на 10 Гб. Как результат OOM Kill (Out Of Memory). Воркер Airflow падает, задача фейлится, соседние легкие задачи тоже умирают, потому что процесс был убит операционной системой. Проблема архитектуры Airflow: PythonOperator выполняет код локально на том воркере, где он запущен. Это значит, что вы ограничены ресурсами одной машины. Пытаться наращивать RAM на воркере Airflow - это тупиковый путь. Решение: Нам нужно вынести исполнение кода за пределы Airflow, оставив за ним только функцию "кнопки пуск" и контроля статуса. Для Python-задач идеальным "внешним процессором" является Dask. В этой статье мы научим Airflow делегировать тяжелые вычисления удаленному кластеру, не меняя при этом привычный Python-стек.  

Архитектура делегирования - Как Airflow использует Dask для масштабирования

  В этой связке Airflow выступает в роли заказчика. Airflow Worker: Запускает задачу. Но вместо того чтобы грузить данные в свою память, он создает легкий объект-клиент. Сетевой вызов: Этот клиент стучится по TCP к планировщику Dask (Scheduler). Удаленное исполнение: Dask забирает инструкцию и данные (из S3), "перемалывает" их на своих мощностях. Ожидание: Airflow-задача висит в ожидании ответа (или мониторит статус), потребляя минимум ресурсов. Для Airflow это выглядит как обычный Python-скрипт, но физически нагрузка уходит на другие серверы.    

Настройка инфраструктуры для масштабирования выполнения задач Airflow

Чтобы Airflow мог управлять Dask-ом, внутри контейнера Airflow должна стоять библиотека dask, s3fs и pandas  нужных версий. Убедитесь, что она есть в вашем Dockerfile или установите её. обновляем dockerfile до: FROM apache/airflow:2.8.1-python3.10 USER root # 1. ОБЪЕДИНЕННАЯ УСТАНОВКА СИСТЕМНЫХ ПАКЕТОВ # gcc, libkrb5-dev... — нужны для компиляции HDFS провайдера (Урок 5) # default-jdk, procps, curl — нужны для работы Spark (Урок 6) RUN apt-get update && apt-get install -y --no-install-recommends gcc libkrb5-dev krb5-user libffi-dev default-jdk procps curl && apt-get autoremove -yqq --purge && apt-get clean && rm -rf /var/lib/apt/lists/* # 2. НАСТРОЙКА JAVA (Урок 6) ENV JAVA_HOME=/usr/lib/jvm/default-java # 3. УСТАНОВКА SPARK-КЛИЕНТА (Урок 6) ENV SPARK_VERSION=3.5.1 ENV HADOOP_VERSION=3 RUN curl -O https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && tar zxf spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz -C /opt/ && rm spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && ln -s /opt/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION} /opt/spark # 4. НАСТРОЙКА ПУТЕЙ SPARK ENV SPARK_HOME=/opt/spark ENV PATH=$PATH:$SPARK_HOME/bin: USER airflow #--- Добавляем ограничения на установку версий совместимых с Airflow и Python ARG AIRFLOW_VERSION=2.8.1 ARG PYTHON_VERSION=3.10 ARG CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt" # 5. УСТАНОВКА ПРОВАЙДЕРОВ # Ставим сразу оба: и для HDFS (чтобы работал DAG из 5 урока), и для Spark (Урок 6) #--- ОБНОВЛЕНИЕ ЗДЕСЬ --- RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}" apache-airflow-providers-apache-hdfs apache-airflow-providers-apache-spark "dask==2023.12.1" "distributed==2023.12.1" s3fs pandas # ПОСЛЕДНЯЯ СТРОКА Добавляем dask, distributed и s3fs (для работы с Yandex S3) только нужной версии Build-им снова docker image airflow-hdfs:2.8.1

Но и это еще не все к сожалению многокомпонентные распределенные системы, часто конфликтуют из-за  совместимости и на стороне выполнения задач ( Dask worker и scheduler). Dockerfile, который мы подготовили, настроен для сборки Airflow  контейнеров, и если вы соберете сейчас имидж с помощью ниже описанного docker-compose.yaml файла, вылезут несовместимости по python библиотекам используемым на Dask узлах поэтому сделаем небольшую сборку для Dask: #--- соберем кастомный имидж для установки Dask контейнеров с треьования по версионности компонент FROM daskdev/dask:2023.12.1 COPY requirements-dask.txt /tmp/requirements-dask.txt RUN pip install --no-cache-dir -r /tmp/requirements-dask.txt && python -c "import s3fs,fsspec,pandas; print('OK', s3fs.__version__, fsspec.__version__, pandas.__version__)" Создадим requirements-dask.txt для сборки нового имиджа под Dask #-- версии подходящие для dask:2023.12.1 fsspec==2023.12.1 s3fs==2023.12.1 aiobotocore==2.7.0 botocore==1.31.64 boto3==1.28.64 pandas Variables c именем 'yandex_creds' # В формате JSON: {"key": "...", "secret": "..."} # ИЛИ (для теста) впишите свои ключи ниже, если не хотите возиться с Variables # aws_key = "ВАШ_ACCESS_KEY" # aws_secret = "ВАШ_SECRET_KEY" # Чтобы не светить ключи в коде, попробуем достать из Environment (если прокинули в docker-compose) import os aws_key = os.getenv("AWS_ACCESS_KEY_ID", "ЗАМЕНИТЕ_НА_КЛЮЧ_ЕСЛИ_НЕТ_ENV") aws_secret = os.getenv("AWS_SECRET_ACCESS_KEY", "ЗАМЕНИТЕ_НА_СЕКРЕТ_ЕСЛИ_НЕТ_ENV") # 4. Отправляем задачу на кластер future = client.submit(heavy_processing_task, BUCKET_NAME, S3_FILE_PATTERN, aws_key, aws_secret) logging.info("Задача отправлена в Dask. Ждем...") # Ждем завершения wait(future) # Получаем результат (путь к файлу) try: result_path = future.result() logging.info(f"Успех! Данные сохранены в: {result_path}") except Exception as e: logging.error(f"Ошибка вычислений в Dask: {e}") raise e client.close() with DAG( dag_id="07.dask_yandex_processing", start_date=datetime(2023, 1, 1), schedule=None, catchup=False, tags= ) as dag: run_on_dask = PythonOperator( task_id="run_on_dask_cluster", python_callable=offload_to_dask ) Не забудьте прописать свои ключи в DAG файл ( конечно с точки зрения это не правильно, но наверняка вы не столкнетесь с кучей ошибок при попытке сохранить результаты из dask worker ов, а это уже совсем другая история которую мы расскажем вам на курсе  ниже по ссылке     А вот и наш результат

Хотя конечно мы все равно получили предупреждения о ращличиях в версиях компонент на Dask кластере, но не критичных для нашего DAG кода.  

Главные ошибки при интеграции Dask (Airflow-Specific)

При такой схеме работы инженеры часто наступают на специфические грабли синхронизации. Проблема синхронизации окружений (Dependency Hell) Airflow "сериализует" (упаковывает) вашу функцию и отправляет её в Dask. Сценарий: Вы используете в функции библиотеку Dask версии 2023.5.0, которая стоит на Airflow. А на Dask-воркере стоит Dask 2026.1.2(latest). Результат: Задача упадет с ошибкой десериализации. Правило Airflow: Образы Docker для Airflow Worker и Dask Worker должны иметь идентичный набор Python-библиотек. Если обновляете requirements.txt в Airflow - обновляйте и в Dask. Давайте поразмышляем  немного над этим. В нашем случае при старте нашего DAG мы сразу получаем ошибку и пытаемся разобраться в причинах

Присмотритесь над выделенным фрагментом - мы  увидим разные версии на клиенте и на сервере ( worker+scheduler) для dask и Python. вспоминаем опции которые мы прописали в dockerfile, когда создавали имидж для поднятия Airflow с библиотеками Dask и имиджа "FROM apache/airflow:2.8.1", но версии утилит Dask и Python, да и других утилит мы не выбирали. Без споров на будущее и с уверенностью что изменение конфигурации имиджа не разрушит совместимость с нашими прошлыми уроками (1-6) и установленными компонентами (Hadoop,Spark) выбираем фиксированную версию исходного имиджа Airflow c Python 3.10  

В dockerfile вносим изменения вместо "FROM apache/airflow:2.8.1" ->"FROM apache/airflow:2.8.1-python3.10" и для установки пакетов Dask, Distributed тоже жестко фиксируем номера версий ("dask==2023.12.1" "distributed==2023.12.1" "dask==2023.12.1" s3fs pandas), чтобы и в docker-compose.yaml при конфигурации dask workers и schedulers использовал не "image: daskdev/dask:latest", а версию максимально близкую и совместимую с клиентом допустим "image: daskdev/dask:2023.12.1".

Пересобираем docker image и проверяем работу DAGа  ( займет чуть больше времени) docker compose down docker build --no-cache -t airflow-hdfs:2.8.1 . docker compose up -d    Ошибка "Вернуть данные в return" Новички часто пишут: return heavy_dataframe. Что происходит: Dask возвращает гигабайты данных по сети обратно в Airflow Client. Airflow пытается записать это в XCom (мета-базу Postgres). Результат: База зависает, Airflow падает. Правило: Задачи на Dask должны читать из S3 и писать в S3. В Airflow возвращаем только пути к файлам или статус (True/False). Сетевая доступность Если Airflow запущен локально, а Dask в Docker (или наоборот), Client(...) не сможет подключиться. В нашем примере все работает, потому что оба сервиса живут в одной сети docker-compose. В реальном проде Airflow и Dask могут быть в разных подсетях Kubernetes, и вам придется настраивать доступы. Исправьте финальные варианты кода Dags  и  конфигурационных файлов и при необходимости сравните с нашими на GitHub где лежит код к Уроку 7.   Роль Cursor в написании "оберток" для масштабирование Airflow c Dask   Код делегирования часто шаблонный. Cursor может сэкономить время. Промпт: Напиши функцию-обертку для PythonOperator Airflow. Функция должна принимать адрес Dask-планировщика и словарь с параметрами. Внутри она должна подключаться к кластеру, запускать переданную функцию processing_logic, ждать результата и логировать прогресс. Обработай возможные ошибки подключения (TimeOut). Итог: Мы научились использовать PythonOperator не как исполнителя, а как контроллер. Теперь Airflow может запускать задачи любой тяжести, делегируя их Dask-кластеру. Мы решили проблему нехватки памяти (OOM) архитектурным способом, не меняя язык программирования. Теперь, когда мы умеем работать с тяжелыми файлами, пришло время поговорить о скорости реакции. В следующей статье мы разберем Event-Driven архитектуру. Мы узнаем, как заставить Airflow запускать DAG не по расписанию, а мгновенно - как только в Kafka прилетело сообщение о событии. Переходим к Kafka и событийной модели?      

Использованные референсы и материалы

Эта статья открывает наш цикл посвященный бесплатному курсу лекций по "Apache Airflow для новичков" и закладывает фундамент для осовения всей темы. Мы разберемся, зачем вообще нужен оркестратор, как Airflow устроен внутри и как его запустить за пять минут, чтобы начать первые эксперименты.  

Архитектура и философия: почему Apache Airflow стал стандартом

  Если вы когда-нибудь настраивали запуск скриптов через обычный системный планировщик задач вроде Cron, то наверняка сталкивались с ситуацией адской цепочки. Это когда один скрипт должен запуститься строго после того, как отработает другой, но первый внезапно упал из-за ошибки в данных, а второй все равно стартовал и наплодил пустых файлов. Или еще хуже: вы вообще не узнали, что что-то сломалось, пока вам не написали разгневанные пользователи отчетов. Apache Airflow появился именно как ответ на этот хаос. Его создали в Airbnb, когда поняли, что управлять сотнями зависимых процессов вручную невозможно. В этой статье мы разберем, почему подход код как конфигурация стал золотым стандартом в дата-инженерии, и как именно работает мозг этого инструмента. Мы пройдем путь от понимания теории до запуска вашей первой рабочей песочницы в Docker.   Проблема. Почему Cron больше не справляется   Представьте стандартный рабочий процесс инженера данных: нужно выкачать лог из Kafka, сохранить его в облако S3, запустить Spark-задачу для обработки и в конце положить результат в аналитическую базу ClickHouse. В мире классических скриптов это выглядит как набор разрозненных записей в текстовом файле планировщика. У такого подхода нет общей памяти и нет понимания связей между задачами. Если Spark-задача зависнет и съест всю память, обычный планировщик просто попытается запустить следующий этап по расписанию, окончательно добив сервер. Основные боли, которые лечит Airflow, кроются в прозрачности и управляемости. В отличие от невидимых скриптов, здесь у вас есть визуальный граф, где каждая стрелка означает жесткую зависимость. Задача Б просто не начнется, пока задача А не вернет статус успеха. К тому же, инструмент берет на себя всю рутину с перезапусками. Если сеть мигнула и выгрузка данных оборвалась, система сама попробует повторить операцию через пять минут, не отвлекая вас от важных дел. Это дает ту самую уверенность в данных, которой так не хватает при использовании самописных костылей.   Философия направленных графов и магия идемпотентности   Прежде чем лезть в настройки, нужно усвоить два термина, на которых держится весь проект. Первый - это DAG (Directed Acyclic Graph) или Направленный Ациклический Граф. Слово направленный означает, что у ваших задач есть четкий вектор движения от начала к концу. Ациклический гарантирует, что в схеме нет петель: задача не может ссылаться сама на себя через цепочку других действий. Это дает уверенность, что процесс когда-нибудь завершится, а не превратится в бесконечный цикл, пожирающий ресурсы.

Второй критически важный термин - идемпотентность. Это свойство означает, что сколько бы раз вы ни запускали одну и ту же задачу с одними и теми же входными данными, результат всегда будет одинаковым. Например, скрипт, который просто добавляет строки в таблицу, не идемпотентен, ведь при повторе он создаст дубликаты. А вот скрипт, который сначала удаляет старые данные за конкретный день, а потом записывает новые - идеальный пример. Airflow буквально заставляет вас писать код именно так, потому что в мире больших данных сбои неизбежны, и возможность безопасно перезапустить упавший пайплайн (Retry) бесценна.   Анатомия Apache Airflow - Как это работает под капотом

Apache Airflow — это не монолитная программа, а распределенная система, где каждый компонент выполняет свою узкую роль. Чтобы понимать, почему задача висит в статусе Queued и не запускается, нужно понимать, как эти компоненты общаются.

База метаданных (Metadata Database) Это «мозг» и память всей системы. Обычно это PostgreSQL или MySQL. Здесь хранится абсолютно всё: настройки подключений, логины пользователей, переменные (Variables) и, самое главное, состояние каждой задачи (Task Instance State). Важно: Компоненты Airflow (Планировщик, Веб-сервер, Воркеры) почти никогда не общаются друг с другом напрямую. Они общаются через чтение и запись в эту базу данных. Если база упадет — Airflow остановится.

Планировщик (Scheduler) Это сердце системы. Это процесс, который постоянно работает в фоне. У него две главные функции:

-

Парсинг: Он постоянно сканирует папку с вашими Python-файлами (dags/), чтобы найти новые или измененные DAG-и.

-

Планирование: Он проверяет в базе данных, наступило ли время запуска для какого-либо DAG-а. Критический нюанс: Сам Планировщик не выполняет ваши задачи (код ETL). Он лишь создает запись в базе данных: «Задачу А пора выполнять, ставлю ей статус Scheduled».

Исполнитель (Executor) Это механизм (модуль внутри Планировщика), который решает, где и как будут выполняться задачи.

-

SequentialExecutor: Простой, однопоточный. Подходит только для отладки.

-

LocalExecutor: Запускает задачи как подпроцессы на той же машине, где работает Планировщик.

-

CeleryExecutor / KubernetesExecutor: Позволяет запускать задачи на удаленных серверах (воркерах). Для нашего старта мы используем LocalExecutor — это золотая середина между простотой настройки и возможностью параллельного запуска.

Веб-сервер (Webserver) То, что вы видите в браузере. Это Flask-приложение, которое просто читает данные из Базы Метаданных и красиво их отрисовывает. Если веб-сервер упадет, ваши пайплайны продолжат работать, просто вы потеряете визуальный контроль.

Такое разделение позволяет системе быть гибкой. Вы можете обновить веб-интерфейс, не останавливая выполнение критически важных расчетов, или заменить базу данных, сохранив всю логику пайплайнов в Python-файлах.

Жизненный цикл задачи - от кода до исполнения  

Понимание этого потока — ключ к успешной отладке (Troubleshooting). Что происходит, когда вы написали код?

-

Parsing: Планировщик читает ваш .py файл. Если там нет синтаксических ошибок, он создает структуру DAG в памяти.

-

Scheduling: Наступает время запуска (start_date + interval). Планировщик создает в БД объект DagRun (запуск графа) и объекты TaskInstance (экземпляры задач) со статусом None.

-

Queuing: Планировщик видит, что зависимости соблюдены, и меняет статус задачи на Scheduled, а затем отправляет её Исполнителю (Executor). Статус меняется на Queued.

-

Execution: Воркер подхватывает задачу, меняет статус на Running и начинает выполнять ваш Python-код.

-

Completion: Если код завершился без ошибок, воркер пишет в БД статус Success. Если было исключение — Failed.

В веб-интерфейсе Airflow этот жизненный цикл отображается цветами: задача меняет цвет с серого (Queued) на салатовый (Running), а затем становится темно-зеленой (Success) или ярко-красной (Failed). Именно за этой "светофором" вы и будете следить в своей повседневной работе, чтобы мгновенно оценивать здоровье пайплайнов.

Таким образом, запуск DAG - это не одномоментное действие, а эстафета, где Планировщик передает палочку Базе данных, а та — Исполнителю. Понимание этой цепочки спасет вам немало нервов при отладке. А теперь, когда мы разобрались с теорией, давайте поднимем эту архитектуру своими руками.  

Практика: поднимаем песочницу Apache Airflow в Docker за пять минут

  Чтобы не тратить часы на установку библиотек и настройку окружения, мы воспользуемся Docker. Это позволит создать изолированную среду, которая не замусорит вашу основную систему. Для старта нам понадобится всего один файл docker-compose.yaml, который опишет все необходимые связи между компонентами. Во избежания ошибок при работе с docker и docker compose требуется проверить и обновить конфигурацию docker compose в вашей OS.

Bash скрипт для обновления docker-compose --version # должно быть выше v1 docker --version # должно быть выше 20.10 #--- если версия удовлетворяет условиям не выполняем код ниже mkdir -p ~/.docker/cli-plugins curl -SL https://github.com/docker/compose/releases/download/v2.25.0/docker-compose-linux-x86_64 -o ~/.docker/cli-plugins/docker-compose chmod +x ~/.docker/cli-plugins/docker-compose docker compose version # проверяем снова # удаляем старую версию which docker-compose sudo rm /usr/bin/docker-compose Ниже приведен пример конфигурации (полный оригинальный вариант docker-compose.yaml доступен здесь ), которая разворачивает связку из базы данных Postgres и самого Airflow в режиме LocalExecutor. Создайте пустую папку, положите туда этот текст и создайте рядом подпапку dags. Обратите внимание на переменную AIRFLOW_UID: она нужна, чтобы файлы, созданные внутри контейнера, принадлежали вашему пользователю, а не руту (root). Примечание: Если у вас не слишком мощный компьютер (ноутбук) и вы не можете выделить 6-8 гигабайт оперативной памяти под Docker - рассмотрите вариант использования  бесплатного (evaluation) аккаунта на одной из Cloud платформ. Обычно каждая из них предлагает Free tier (AWS) или какое то количество денег (Yandex Cloud) . В этом случае просто создайте вирутальную машину с одной из стандартных операционных  систем ( Ubuntu 24) и установите Docker.   # Создаем папку проекта mkdir airflow-course && cd airflow-course mkdir -p dags logs plugins # Скачиваем docker-compose файл для установки Apache Airflow 2.8.1 c LocalExecutor curl -LfO 'https://raw.githubusercontent.com/BigDataSchoolRU/articles_code/refs/heads/main/airflow_course/article01/docker-compose.yaml' -o docker-compose.yaml #-- для получения AIRFLOW UID создаем файл .env для текущего пользователя, и теперь мы можем получить доступ к папке с DAGs  echo "AIRFLOW_UID=$(id -u)" > .env # Инициализация и запуск docker compose up airflow-init docker compose up -d Инициализируем базу данных postgres

После того как файл готов, выполните команду docker compose up -d. Через пару минут интерфейс будет доступен по адресу localhost:8080 с логином и паролем admin. Теперь у вас есть полноценная лаборатория для опытов, где можно безбоязненно ломать и чинить любые процессы, просто перезапуская контейнеры.

Для проверки запуска сервисов используйте # ---- список запущенныз docker container docker ps #---- проверка логов сервера airflow ( на выбор) docker compose logs scheduler

После завершения упражнений с Apache Airflow погасите его командой docker compose down Все примеры кода, которые мы показывали на данном уроке и последующих (см. ссылки под статьей) вы можете скачать на нашем Репо GitHub    BashOperator против PythonOperator   В Airflow вся работа выполняется Операторами. Это кирпичики, из которых строится DAG. Когда база настроена, пора переходить к написанию кода. Самый простой выбор для новичка - это BashOperator и PythonOperator. Давайте сравним их на примере простой задачи: нужно создать папку с текущей датой, чтобы понять, какой инструмент лучше подходит для ваших сценариев. BashOperator - это способ выполнить любую команду в терминале. Если вы привыкли писать скрипты на bash или использовать готовые утилиты вроде curl или rsync, этот оператор станет вашим лучшим другом. Он очень быстрый и потребляет минимум ресурсов, потому что просто пробрасывает команду в операционную систему. Однако его трудно отлаживать, если логика становится сложнее одного предложения, так как передавать данные между задачами через bash - то еще удовольствие. PythonOperator, напротив, дает вам всю мощь языка программирования. Вы можете импортировать любые библиотеки, работать с API или делать сложные вычисления прямо в коде задачи. Это гибко, но требует больше ресурсов сервера, так как каждый запуск задачи порождает новый тяжеловесный процесс Python. В таблице ниже мы собрали ключевые различия, чтобы вам было проще ориентироваться. Характеристика BashOperator PythonOperator Когда использовать   Системные команды, перемещение файлов, запуск бинарников   Сложная логика, работа с API, обработка данных в Pandas Скорость запуска   Почти мгновенно   Зависит от объема импортируемых библиотек Читаемость   Хорошая для коротких однострочников   Отличная для разработчиков Python Передача данных   Через временные файлы или переменные окружения   Нативная через механизм XCom (Cross-Communication)   Выбор часто зависит от того, где живет логика. Если у вас уже есть готовый скрипт на 500 строк, проще запустить его через BashOperator. Если же вы строите новый процесс с нуля, PythonOperator обеспечит лучшую интеграцию с экосистемой Apache Airflow.   Чтение логов: как понять что пошло не так   Когда задача запускается, Airflow начинает подробно записывать каждое свое действие. Умение читать эти логи - половина успеха инженера данных. В интерфейсе вы всегда можете кликнуть на квадратик задачи и нажать кнопку Log. Там вы увидите не только вывод вашего кода, но и служебную информацию от самой системы, которая часто оказывается важнее самого кода.

В успешном логе вы увидите пометку о начале работы процесса (PID), информацию о том, какой Executor взял задачу, и заветную строчку Marking task as SUCCESS. Если же задача упала, ищите ключевые слова Traceback или ERROR. Airflow честно покажет вам строку кода, на которой произошел сбой. Часто бывает, что проблема не в логике, а в окружении: например, вы забыли установить нужную Python-библиотеку внутри Docker-контейнера, и система просто не может сделать import. Помните, что логи планировщика и логи конкретной задачи - это разные вещи. Если задача даже не началась, ответ нужно искать в логах контейнера scheduler.   Частые затыки и как из них выбираться   Даже с готовой инструкцией легко совершить ошибки, которые ставят в тупик. Самая частая из них связана с параметром start_date. Новички часто ставят текущее время и удивляются, почему ничего не происходит. Логика Airflow такова, что он запускает задачу только после завершения первого интервала времени. Если вы хотите, чтобы DAG запустился прямо сейчас, ставьте дату начала на один день в прошлом. Это классика, через которую проходят абсолютно все. Еще один камень преткновения - это отсутствие динамики в интерфейсе. Если вы изменили код в файле, а в браузере ничего не поменялось, проверьте вкладку Import Errors в самом верху страницы. Скорее всего, где-то пропущена запятая или нарушены отступы. Планировщик просто не может прочитать битый файл и продолжает показывать вам старую версию из своей памяти, чтобы не ломать работающую систему. Если там пусто, зайдите внутрь контейнера планировщика (docker exec -it bash) и выполните airflow dags list-import-errors. Или просто посмотрите логи планировщика в терминале. Наконец, не забывайте про ресурсы вашего компьютера. Если вы запускаете тяжелую обработку данных внутри PythonOperator на слабом ноутбуке, Docker-контейнер может просто молча перезагрузиться или зависнуть. Всегда следите за тем, сколько оперативной памяти выделено вашему Docker Desktop (минимум 4 ГБ для комфортной работы), иначе даже самый правильный код будет вылетать без внятных объяснений.Проверьте, запущен ли сервис scheduler. В нашем Docker-compose это отдельный контейнер. Если он упал (например, из-за нехватки памяти OOM), задачи никогда не перейдут в статус Running.  

Помощь Cursor в документировании и отладке

В прошлой статье мы научили Airflow работать со структурированными данными в Postgres. Но в мире Big Data базы данных - это лишь верхушка айсберга. Основная масса данных (логи, картинки, JSON-выгрузки, бэкапы) хранится в виде файлов. Хранить эти файлы на локальном диске сервера, где крутится Airflow - плохая идея. Диск не резиновый.  Воркеры эфемерны. Если вы используете Celery или Kubernetes, воркер может умереть и возродиться на другом сервере, и ваш локальный файл исчезнет. Решение - вынести хранение данных во внешнее объектное хранилище, совместимое с протоколом S3. В этой статье мы превратим наш Airflow в полноценный ETL-инструмент: научим его забирать данные, сохранять их в локальный S3 (MinIO) и переключаться на облачный Yandex Object Storage одной строчкой конфига.  

Что такое S3 и почему это не только Amazon

  Изначально S3 (Simple Storage Service) придумали в Amazon. Но протокол оказался настолько удачным, что стал мировым стандартом. Сегодня, когда говорят "S3", чаще имеют в виду не конкретный сервис Amazon, а протокол общения. Если приложение умеет работать с S3, оно автоматически умеет работать с: MinIO (ваш личный S3 на сервере) Yandex Object Storage Google Cloud Storage Ceph и другими системами Для Airflow нет разницы, куда писать. Ему нужны только три вещи: Endpoint (адрес), Access Key (логин) и Secret Key (пароль).   Поднимаем свой S3 (MinIO) в Docker   Чтобы тренироваться бесплатно и локально, добавим в наш docker-compose.yaml сервис MinIO. Это легкий и быстрый S3-сервер. Добавьте этот блок в секцию services (рядом с postgres и redis): minio: image: minio/minio:latest ports: - "9000:9000" # API порт (для Airflow) - "9001:9001" # Веб-консоль (для вас) environment: - MINIO_ROOT_USER=minioadmin - MINIO_ROOT_PASSWORD=minioadmin command: server /data --console-address ":9001" healthcheck: test: interval: 30s timeout: 20s retries: 3 volumes: - minio_data:/data volumes: # Не забудьте объявить том в конце файла minio_data: Проверьте docker-compose.yaml файл на валидность с помощью docker compose config и перезапустите кластер:   docker compose config #-- так мы будем уверенны что не будет хвостов на нашем стенде docker-compose down  --remove-orphans docker compose up -d --force-recreate Теперь зайдите в браузере на localhost:9001. Логин/пароль: minioadmin / minioadmin.

Важно: Сразу создайте там бакет (аналог папки) с именем airflow-bucket.     Настраиваем Connection в Airflow   Самый сложный момент для новичков - правильно настроить соединение. Airflow использует библиотеку boto3 для работы с S3, и она по умолчанию пытается стучаться в Amazon. Нам нужно переубедить её. Вариант А: Как настроить Connection Apache Airflow для MinIO (Локально) Зайдите в Airflow -> Admin -> Connections. Создайте новое соединение: Conn Id minio_s3 Если такого типа нет, нужно установить провайдер: pip install apache-airflow-providers-amazon, но в стандартном образе он обычно есть Conn Type Amazon Web Services AWS Access Key ID minioadmin AWS Secret Access Key minioadmin Extra {"endpoint_url": "http://minio:9000"} Это самое важное поле! Нам нужно указать, что адрес сервера - не амазон, а наш контейнер Слава богу это YAML -просто JSON  можно писать в строчку ВАЖНО: двойные кавычки для поле extra Обратите внимание: мы используем имя сервиса minio из docker-compose, так как Airflow будет стучаться к нему изнутри сети Docker.   Вариант Б: Как настроить Connection Apache Airflow для Yandex Object Storage (Продакшен) Если у вас есть облако в Яндексе, процесс почти идентичен. В консоли Y.Cloud создайте Сервисный аккаунт. Дайте ему роль storage.editor. Создайте для него Статический ключ доступа. Вы получите идентификатор ключа и секретный ключ. В Airflow создайте соединение yandex_s3 Conn Type Amazon Web Services Login (Access Key) Ваш ключ из Y.Cloud Password (Secret Key) Ваш секрет из Y.Cloud Extra {"endpoint_url": "https://storage.yandexcloud.net"}

Видите? Разница только в endpoint_url и ключах. Код DAG-а менять не придется!    

Практика: Пишем данные в S3

  Давайте решим классическую задачу. У нас есть данные в Postgres (из прошлой статьи), и нам нужно выгрузить их в CSV-файл и сохранить в "Озеро Данных" (S3) для дальнейшей аналитики. Мы будем использовать S3Hook. Это универсальный инструмент для работы с объектным хранилищем. Создайте файл dags/postgres_to_s3.py: from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.postgres.hooks.postgres import PostgresHook from airflow.providers.amazon.aws.hooks.s3 import S3Hook from datetime import datetime import csv import os # Название бакета (должен быть создан заранее!) BUCKET_NAME = "airflow-bucket" # Имя файла с датой запуска KEY_NAME = "users_export_{{ ds }}.csv" def export_postgres_to_s3(ds, **kwargs): # 1. Забираем данные из Postgres pg_hook = PostgresHook(postgres_conn_id="my_dwh") connection = pg_hook.get_conn() cursor = connection.cursor() cursor.execute("SELECT * FROM users") results = cursor.fetchall() # 2. Сохраняем во временный локальный файл # Важно: /tmp/ очищается, не засоряя диск local_filename = f"/tmp/users_{ds}.csv" with open(local_filename, 'w') as f: csv_writer = csv.writer(f) csv_writer.writerow() # Заголовки csv_writer.writerows(results) print(f"Данные выгружены локально: {local_filename}") # 3. Загружаем в S3 (MinIO или Yandex) # Используем conn_id, который мы настроили (minio_s3 или yandex_s3) s3_hook = S3Hook(aws_conn_id="minio_s3") s3_hook.load_file( filename=local_filename, key=KEY_NAME, # Имя файла в облаке bucket_name=BUCKET_NAME, replace=True # Перезаписывать, если файл уже есть (Идемпотентность!) ) print(f"Файл успешно загружен в S3: {BUCKET_NAME}/{KEY_NAME}") # 4. Убираем за собой (удаляем локальный файл) os.remove(local_filename) with DAG( dag_id="export_to_datalake", start_date=datetime(2023, 1, 1), schedule=None, catchup=False ) as dag: upload_task = PythonOperator( task_id="upload_to_s3", python_callable=export_postgres_to_s3 )

Разбор кода - Локальный буфер: Мы сначала пишем файл на диск (/tmp), а потом грузим его. Почему не сразу потоком? Для новичка файл надежнее, его можно проверить, если загрузка упадет. - replace=True: Это обеспечивает идемпотентность. Если вы запустите DAG дважды за один день, старый файл просто перезапишется новым. Данные не задублируются. - Универсальность: Если завтра вы решите переехать с MinIO на Yandex, вам нужно будет поменять только одну строчку: aws_conn_id="minio_s3" на aws_conn_id="yandex_s3".  

Troubleshooting: Почему не летит?

  Работа с S3 полна нюансов сетевого взаимодействия. Ошибка 1 EndpointConnectionError: Could not connect to the endpoint URL Причина: Airflow не видит MinIO. ( Мы намерянно поправим порт для MinIO на 9009 - на котором глушняк :-)

Лечение: Проверьте поле Extra в Connection. Если вы написали localhost:9000, это ошибка (Airflow будет искать MinIO внутри своего контейнера). Должно быть http://minio:9000 (имя сервиса из docker-compose). Ошибка 2: 403 Forbidden Причина: Неверные ключи доступа или (в случае Yandex) у сервисного аккаунта нет прав на запись в бакет. Лечение: Проверьте Access/Secret Key. В Yandex Cloud убедитесь, что роль storage.editor назначена именно на тот каталог (folder), где лежит бакет. Ошибка 3: The specified bucket does not exist Причина: S3Hook (обычно) не создает бакеты сам. Мы переключили в dags/postgres_to_s3.py подключение с MinIO на Yandex Cloud, но забыли создать там bucket и вот результат вторая ошибка

  Лечение: Зайдите в консоль MinIO (localhost:9001) или Yandex Cloud и создайте бакет airflow-bucket вручную перед запуском DAG. Исправьте финальные варианты и сравните с нашими на GitHub где лежит код наших уроков.    

Помощь Cursor: Работа с boto3

  S3 API огромен. Запоминать все методы S3Hook не нужно. Сценарий 1: Проверка наличия файла. Попросите Cursor написать код и потом совместить его с кодом вашего дага: "Напиши код для Airflow PythonOperator, который использует S3Hook для проверки: если файл 'flag.txt' существует в бакете, то мы идем дальше, иначе падаем с ошибкой. Используй conn_id='yandex_s3'."

Созданный Cursor AI код check_s3_flag_dag.py и post_new_s3.py собранный из вашего и сгенерированного кода, с описанием вы можете посмотреть на github/article04

Сценарий 2: Генерация JSON-конфига Если вы путаетесь в JSON для поля Extra, напишите: "Сгенерируй JSON для поля Extra в Airflow Connection, чтобы подключиться к Yandex Object Storage. Endpoint должен быть https://storage.yandexcloud.net." Итог: Теперь наш Airflow - это не вещь в себе. Он умеет забирать данные из продуктовых баз и складывать их в надежное хранилище. Мы построили простейший Data Lake. Но что делать с этими файлами дальше? Просто лежать они не приносят пользы. В мире Big Data для обработки огромных массивов файлов используют Hadoop и HDFS. В следующей статье мы коснемся "тяжелой артиллерии". Разберем, как Airflow управляет процессами в экосистеме Hadoop и зачем нам нужен WebHDFSSensor, даже если мы не пишем на Java. Готовы погрузиться в мир Hadoop?  

Использованные референсы и материалы

Мы построили пайплайн, где данные забираются из базы и бережно складываются в HDFS. Теперь они лежат там мертвым грузом. Чтобы превратить сырые CSV в полезные отчеты, их нужно обработать: отфильтровать, агрегировать, джойнить. Делать это внутри самого Airflow (через PythonOperator и Pandas) - плохая идея если: Память: Если файл весит 100 ГБ, ваш воркер Airflow просто лопнет (OOM Kill). Скорость: Pandas работает на одном ядре. Spark распределяет задачу на сотни ядер. Airflow здесь выступает только как кнопка "Пуск". Он запускает задачу на кластере Spark и смиренно ждет, пока "большой брат" закончит работу. Для этого используется SparkSubmitOperator.  

Главная проблема - Airflow не умеет заходить в Spark "из коробки"

Здесь новички ломают копья. SparkSubmitOperator - это, по сути, обертка над консольной командой spark-submit. Чтобы этот оператор сработал, на машине (или в контейнере), где крутится воркер Airflow, должны быть физически установлены: Java (OpenJDK) — потому что Spark работает на JVM. Клиент Spark — набор бинарных файлов, чтобы отправить команду кластеру. Стандартный образ apache/airflow, который мы использовали до этого, не содержит Java. Если вы попробуете запустить Spark-задачу сейчас, вы получите ошибку JAVA_HOME not set или spark-submit not found. Помните как мы с вами костылизировали с Apache Hadoop и AirFlow на прошлом уроке?       Шаг 1. Модернизируем Docker-образ для поддержки Java и Apache Spark   Нам придется снова немного "запачкать руки" и пересоздать свой образ Airflow, который мы с вами использовали на прошлом уроке ( поддержка Hadoop provider) и добавить туда Spark с поддержкой Java. Не пугайтесь, это стандартная практика. Создайте(или отредактируйте существующий)  файл Dockerfile в корне проекта (рядом с docker-compose): FROM apache/airflow:2.8.1 USER root # 1. ОБЪЕДИНЕННАЯ УСТАНОВКА СИСТЕМНЫХ ПАКЕТОВ # gcc, libkrb5-dev... — нужны для компиляции HDFS провайдера (Урок 5) # default-jdk, procps, curl — нужны для работы Spark (Урок 6) RUN apt-get update && apt-get install -y --no-install-recommends gcc libkrb5-dev krb5-user libffi-dev default-jdk procps curl && apt-get autoremove -yqq --purge && apt-get clean && rm -rf /var/lib/apt/lists/* # 2. НАСТРОЙКА JAVA (Урок 6) ENV JAVA_HOME=/usr/lib/jvm/default-java export JAVA_HOME # 3. УСТАНОВКА SPARK-КЛИЕНТА (Урок 6) ENV SPARK_VERSION=3.5.1 ENV HADOOP_VERSION=3 RUN curl -O https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && tar zxf spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz -C /opt/ && rm spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && ln -s /opt/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION} /opt/spark # 4. НАСТРОЙКА ПУТЕЙ SPARK ENV SPARK_HOME=/opt/spark ENV PATH=$PATH:$SPARK_HOME/bin USER airflow # 5. УСТАНОВКА ПРОВАЙДЕРОВ # Ставим сразу оба: и для HDFS (чтобы работал DAG из 5 урока), и для Spark RUN pip install --no-cache-dir apache-airflow-providers-apache-hdfs apache-airflow-providers-apache-spark Пересоздаем наш образ "франкенштейн" состоящий теперь из бинарников Hadoop, Spark и провайдеров для них в Airflow вместе с Java - такова цена интеграции. #--- и пересобираем его с нуля для уверенности - процесс может занять 5-10 минут docker build --no-cache -t airflow-hdfs:2.8.1 . #--- почистим старые images docker image ls #для проверки docker image prune docker image ls #для проверки #--- Уже запущенные контейнеры не обновят автоматически используемый image без опции --build docker compose up -d --build

  Теперь нужно сказать docker-compose, чтобы он использовал вновь созданный файл image 'airflow-hdfs:2.8.1', а не готовый образ из интернета. Измените секцию x-airflow-common в docker-compose.yaml добавив новые переменные и volumes для Spark Jobs: x-airflow-common: &airflow-common image: airflow-hdfs:2.8.1 # Изменяем используемый имидж # build: . environment: # ... старые переменные ... JAVA_HOME: /usr/lib/jvm/default-java # Добавляем новые переменные для Spark SPARK_HOME: /opt/spark volumes: # --- НОВЫЙ ТОМ ДЛЯ УРОКА 6 (Скрипты Spark) --- - ${AIRFLOW_PROJ_DIR:-.}/jobs:/opt/airflow/jobs   Шаг 2. Добавляем Spark-кластер   Нам нужен сам Spark, который будет выполнять работу. Добавим мастер и один воркер в docker-compose.yaml (в секцию services): # Урок 6 Spark jobs - кластер для Airflow tasks spark-master: image: apache/spark:3.5.1 container_name: spark-master hostname: spark-master profiles: - spark environment: - SPARK_NO_DAEMONIZE=true - SPARK_MASTER_HOST=spark-master ports: - "9090:8080" # Web UI (сместили на 9090, чтобы не конфликтовал с Airflow) - "7077:7077" # Мастер-порт command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master spark-worker: image: apache/spark:3.5.1 container_name: spark-worker profiles: - spark environment: - SPARK_NO_DAEMONIZE=true - SPARK_WORKER_CORES=1 - SPARK_WORKER_MEMORY=1G depends_on: - spark-master command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker spark://spark-master:7077 Не забудьте включить hadoop & datanode из прошлого примера или применить --profile spark.  Теперь выполняем команду пересборки и запуска: docker-compose up -d --build

Зайдите на localhost:9090. Вы должны увидеть интерфейс Spark Master с одним живым воркером (Alive Workers: 1).

  Шаг 3. Настраиваем Connection Airflow должен знать, где живет Spark Master. Admin -> Connections. Новое соединение: Conn Id my_spark_conn Conn Type Spark Host spark://spark-master Port 7077 Extra ( или Deploy mode) {"deploy-mode": "client"} или cluster, но для Docker проще client, чтобы видеть логи сразу   Практика: PySpark Job для обработки данных   В прошлой статье мы положили файл users_{ds}.csv в HDFS. Давайте напишем скрипт на PySpark, который читает этот файл, считает распределение пользователей по датам и сохраняет отчет обратно в HDFS (или выводит в консоль). Создайте папку jobs рядом с dags и положите туда скрипт user_analytics.py: import sys from pyspark.sql import SparkSession from pyspark.sql.functions import col, count def main(input_path, output_path): # Создаем сессию spark = SparkSession.builder .appName("AirflowUserAnalytics") .getOrCreate() print(f"Reading from: {input_path}") # Читаем CSV. Так как хедер мы писали сами, указываем header=True df = spark.read.option("header", "true").csv(input_path) # Простая аналитика: сколько регистраций в каждую дату report = df.groupBy("date").agg(count("*").alias("total_users")) report.show() # Сохраняем результат (в формате Parquet, это стандарт для Big Data) # report.write.mode("overwrite").parquet(output_path) spark.stop() if __name__ == "__main__": # Аргументы передаются из Airflow if len(sys.argv) != 3: print("Usage: user_analytics.py ") sys.exit(-1) main(sys.argv, sys.argv)   Шаг 4. AirFlow DAG со SparkSubmitOperator    Теперь самое главное - связать всё вместе. Код DAG-а (dags/process_spark.py): from airflow import DAG from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator from datetime import datetime # Путь к HDFS, куда мы писали в прошлой статье # Обратите внимание: spark внутри Docker-сети может обращаться к namenode HDFS_INPUT = "hdfs://namenode:8020/user/airflow/backup/users_{{ ds }}.csv" HDFS_OUTPUT = "hdfs://namenode:8020/user/airflow/reports/daily_{{ ds }}" with DAG( dag_id="06.spark_processing", start_date=datetime(2023, 1, 1), schedule=None, catchup=False ) as dag: process_task = SparkSubmitOperator( task_id="run_spark_job", conn_id="my_spark_conn", # Путь к скрипту внутри контейнера AIRFLOW application="/opt/airflow/jobs/user_analytics.py", # Аргументы для скрипта (input и output) application_args=, # Конфигурация ресурсов (сколько отдать Спарку) conf={ "spark.driver.memory": "512m", "spark.executor.memory": "512m" }, # Важно! Указываем пакеты, если нужно работать с S3 или другими системами # packages="org.apache.hadoop:hadoop-aws:3.2.0", verbose=True ) Как это работает - объяснялки Airflow (Worker) берет параметры из оператора Он формирует длинную команду: /opt/spark/bin/spark-submit --master spark://spark-master:7077 ... user_analytics.py .... Эта команда запускается в контейнере Airflow Клиент Spark связывается с мастером, передает ему код Мастер распределяет задачу на spark-worker Воркер читает данные из HDFS, считает и пишет результат Airflow видит, что процесс spark-submit завершился с кодом 0, и красит задачу в зеленый цвет

  Давайте попробуем?  

Конечно проверить финальный результат из карманного Hadoop и в паркетном формате  тот еще дополнительный квест, но он того стоил ( через docker exec -it скопировать данные с hdfs на локальную систему namenode, потом через docker cp скопировать все с namenode локально и уже финально установив pandas и fastparquet прочитать все с использованием простого питон скрипта). По традиции все финальные версии файлов вы сможете найти у нас на Git    

Troubleshooting: Почему Spark падает?

Запуск Spark в контейнерах - это минное поле. Ошибка 1: JAVA_HOME is not set Причина: Вы не пересобрали Docker-образ или забыли ENV JAVA_HOME в Dockerfile. Лечение: Проверьте docker exec -it java -version. Если команды нет - пересобирайте образ. Ошибка 2: Connection refused к HDFS внутри Spark-джобы Причина: Spark Worker (отдельный контейнер) не может достучаться до Namenode. Лечение: Убедитесь, что все контейнеры (spark-worker, namenode) находятся в одной сети default (docker-compose делает это автоматически, но если вы запускали их разными файлами - будут проблемы). Ошибка 3: Driver runs, but Executor fails Причина: Часто это нехватка памяти. Spark по умолчанию может просить 1ГБ на экзекьютор, а выделили вы Докеру всего 2ГБ на всё. Лечение: Явно занижайте память в параметре conf: "spark.executor.memory": "512m".  

Помощь Cursor: Пишем PySpark код для DAG Airflow

  Если вы не помните синтаксис DataFrame API, Cursor сделает это за вас. "Напиши скрипт на PySpark, который читает Parquet-файлы из папки /data/input, фильтрует пользователей старше 18 лет, группирует их по городам и сортирует по убыванию количества. Результат сохрани в CSV."   Промпт для дебага: "Вот лог ошибки SparkSubmitOperator: . Объясни, почему Airflow не может найти класс org.apache.hadoop.fs.s3a.S3AFileSystem и какой пакет нужно добавить в spark-submit." Итог: Мы преодолели самый сложный барьер интеграции - настроили запуск Java-приложений из Python-оркестратора. Теперь Airflow управляет мощнейшим кластером обработки данных Apache Spark. Но Spark - это тяжело, долго и требует много памяти. Всегда ли нам нужен такой монстр если тем более у Вас его еще нет и не кому его приручить? В следующей статье мы рассмотрим легкую, питоническую альтернативу - Dask. Мы узнаем, как масштабировать Pandas-код на несколько серверов без установки Java и мучений с spark-submit. Переходим к Dask?    

Использованные референсы и материалы

В мире Big Data технологии меняются с бешеной скоростью, но слон (Hadoop) все еще в комнате. Несмотря на популярность облачных S3-хранилищ, распределенная файловая система HDFS остается стандартом де-факто для многих корпоративных хранилищ Data Lake и on-premise кластеров. Даже если вы не пишете MapReduce-задачи на Java, ваш Airflow, скорее всего, должен будет уметь класть файлы в HDFS или забирать их оттуда для Spark-джоб. В этой статье мы продолжим наш сквозной пример. В прошлый раз мы научились выгружать данные из Postgres и сохранять их в S3. Теперь представим, что аналитический отдел требует эти данные к себе в кластер Hadoop для построения витрин. Наша задача - автоматизировать этот переезд, используя новые инструменты Airflow, Сенсоры и WebHDFS.  

Проблема "Java Hell". Почему WebHDFS лучше нативного клиента?

  Когда новичок пытается подключиться к Hadoop из Python, он обычно ищет библиотеку, которая работает по нативному протоколу hdfs://. И тут начинается боль. Чтобы это заработало, на каждом воркере Airflow должны стоять бинарники Hadoop, правильно настроенные переменные окружения JAVA_HOME и HADOOP_CONF_DIR. Это превращает поддержку Docker-образов в ад. Airflow предлагает более изящное решение - WebHDFS. Это REST API над файловой системой Hadoop. Вместо сложного бинарного протокола мы просто отправляем HTTP-запросы (PUT, GET, LIST). Плюсы: Не нужно ставить Java и Hadoop-клиент на воркеры Airflow. Работает через обычный Python requests. Минусы: Чуть медленнее при передаче терабайтов данных (но для оркестрации это редкость). Для Airflow это идеальный выбор. Мы будем использовать WebHDFSHook.    

Обновляем инфраструктуру - добавляем карманный Hadoop

  Чтобы наш пример был рабочим, нам нужно добавить HDFS в наш docker-compose.yaml. (скачайте чистый рабочий из 2 урока или почистите предыдущий от MinIO для практики) Полноценный кластер съест всю память, поэтому мы используем облегченную версию с одним узлом (Namenode + Datanode в одном флаконе). При настройке подключения HDFS в Airflow на практике часто всплывает набор типичных проблем, которые не очевидны из документации: - во-первых, Airflow не работает с HDFS по бинарному RPC-порту (8020) и требует WebHDFS по HTTP (обычно 9870), поэтому необходимо корректно пробросить именно HTTP-порт NameNode и убедиться, что WebHDFS реально доступен внутри docker-сети; - во-вторых, для работы с HDFS в Airflow требуется отдельный provider apache-airflow-providers-apache-hdfs, который не входит в базовый образ Airflow и попытки установить его через _PIP_ADDITIONAL_REQUIREMENTS на старте контейнера часто приводят к падению airflow-init, конфликтам зависимостей и нестабильному поведению. Правильная последовательность действий выглядит так - сначала на стороне Hadoop настраивается fs.defaultFS, включается WebHDFS и пробрасывается порт 9870. Добавьте этот сервис в ваш файл конфигурации: hadoop: image: bde2020/hadoop-namenode:2.0.0-hadoop3.2.1-java8 container_name: namenode environment: - CLUSTER_NAME=test - HDFS_CONF_dfs_permissions_enabled=false # Отключаем проверку прав для простоты - CORE_CONF_fs_defaultFS=hdfs://namenode:8020 ports: - "9870:9870" # Порт WebHDFS (важно!)/ NameNode UI - "8020:8020" # Бинарный порт HDFS RPC (нужен Spark, не Airflow) healthcheck: test: interval: 30s timeout: 20s retries: 3 datanode: profiles: - storage image: bde2020/hadoop-datanode:2.0.0-hadoop3.2.1-java8   container_name: datanode environment: - CORE_CONF_fs_defaultFS=hdfs://namenode:8020 - HDFS_CONF_dfs_permissions_enabled=false      - SERVICE_PRECONDITION=namenode:9870 namenode:8020 depends_on: hadoop: condition: service_healthy ports: - "9864:9864" volumes: - datanode:/hadoop/dfs/data # Не забудьте объявить том в конце файла volumes: datanode: Перезапустите кластер: docker compose up -d. Теперь по адресу localhost:9870 доступен веб-интерфейс Hadoop. Вы можете зайти туда и через меню Utilities -> Browse the file system посмотреть, какие файлы лежат на диске.

Можете также проверить его командой   #--- должен быть один живой datanode docker exec -it namenode hdfs dfsadmin -report #--- на крайний случай создайте все необходимые папки сразу docker exec -it namenode hdfs dfs -mkdir -p /user/airflow/backup/ Итак минипакетный Hadoop кластер с HDFS для наших опытов с интеграцией с Apache Airflow готов.   Настройка Connection Airflow + Hadoop   Настроим соединение, по аналогии с Postgres и S3. Перейдите в Admin -> Connections. Создайте новое соединение: Conn Id my_hdfs_conn Conn Type HDFS Apache WebHDFS Host namenode имя сервиса из docker-compose Port 9870 стандартный порт WebHDFS для Hadoop 3.x. Для старых версий это 50070 Login root или hadoop, зависит от образа, в нашем случае root работает, так как мы отключили проверку прав Но скорее всего у вас не будет возможности добавить WebHDFS, потому что - УПС, по умолчанию его нет. Поэтому правильная последовательность действий выглядит так: - сначала на стороне Hadoop настраивается fs.defaultFS, включается WebHDFS и пробрасывается порт 9870(мы это уже сделали!); - затем для Airflow собирается custom Docker image, в который провайдер HDFS устанавливается на этапе docker build (а не во время запуска); Приступим к костылизации. Создаем dockerfile

FROM apache/airflow:2.8.1 USER airflow RUN pip install --no-cache-dir apache-airflow-providers-apache-hdfs #--- и собираем его - процесс может занять 5-10 минут docker build -t airflow-hdfs:2.8.1 . #--- проверьте доступные теперь имиджи командой docker image ls

  - после этого все сервисы Airflow (webserver, scheduler, worker, init) используют этот единый образ через x-airflow-common; Меняем настройки в docker-compose.yaml файле, чтобы вместо чистого образа airflow_2.8.1 использовался ваш 'костыль с Apache HDFS провайдером'

- и только в финале в UI Airflow создаётся connection типа WebHDFS с указанием host, порта и proxy_user. Такой подход убирает скрытые ошибки, делает окружение воспроизводимым и избавляет от «магии» установки зависимостей при старте контейнеров. Примечание: Если вы скачали наш финальный файл для запуска Airflow c Hadoop и HDFS провайдером заодно обратите внимание на опцию profiles в docker-compose секции уроков 4 MinIO и 5 Hadoop  - мы решили не удалять дельты конфигов наших предыдущих уроков из файла а просто запускать их под разными профилями

Ниже полезные команды #---- запуск кластера Apache Airflow в CeleryExecutor mode docker compose up -d #--- запуск дополнительно S3 MinIO кластера profile cloud docker compose --profile cloud up -d #--- запуск дополнительно Hadoop HDFS кластера profile storage docker compose --profile storage up -d #--- выключить все docker compose --profile '*' down  

Новая концепция: Сенсоры (Sensors)

Прежде чем перекладывать файл, нам нужно убедиться, что он вообще появился в S3. В прошлых статьях мы запускали задачи по цепочке: Сделай А -> Сделай Б. Но что, если файл в S3 кладет не Airflow, а какая-то внешняя система? Или предыдущий DAG задержался? Если мы просто попытаемся скачать несуществующий файл, задача упадет с ошибкой. Можно написать цикл с time.sleep(), но это костыль. В Airflow для этого есть Сенсоры. Sensor - это специальный тип оператора, задача которого - ждать. Он запускается, проверяет условие (есть ли файл? наступило ли время? появилась ли запись в базе?), и если нет - засыпает (переходит в статус up_for_reschedule), освобождая слот воркера. В нашем сквозном примере мы будем использовать S3KeySensor.     Практика: DAG "S3 to HDFS" пример интеграции AirFlow   Напишем DAG, который будет мониторить появление файла в каталоге HDFS или в S3 бакете и перекладывает его в другую папку: Ждет появления файла users_export_{ds}.csv в S3 (который мы создали в прошлой статье). Скачивает его во временную папку. Загружает его в HDFS по пути /user/airflow/backup/. Создайте файл dags/s3_to_hdfs_migration.py: from airflow import DAG from airflow.operators.python import PythonOperator from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.providers.amazon.aws.hooks.s3 import S3Hook from airflow.providers.apache.hdfs.hooks.webhdfs import WebHDFSHook from datetime import datetime import os BUCKET_NAME = "airflow-course" # Шаблоны оставляем здесь, но использовать их будем через параметры оператора S3_KEY_TEMPLATE = "users_export_{{ ds }}.csv" HDFS_PATH_TEMPLATE = "/user/airflow/backup/users_{{ ds }}.csv" # 1. Добавляем аргументы s3_key и hdfs_dest в функцию def transfer_s3_to_hdfs(s3_key, hdfs_dest, **kwargs): print(f"Обрабатываем файлы. Источник: {s3_key}, Назначение: {hdfs_dest}") # 1. Скачиваем файл из S3 s3_hook = S3Hook(aws_conn_id="Yandex_s3") local_filename = s3_hook.download_file( key=s3_key, # move_data   Разбор нюансов DAG-а

Перед запуском обратите внимание на инфраструктуру: так как мы скачиваем данные из S3 (Yandex), убедитесь, что у вас запущен соответствующий профиль: docker compose --profile cloud up -d.

Экономия ресурсов с mode="reschedule" Это критически важный параметр для сенсоров.

-

mode="poke" (по умолчанию): Сенсор занимает целый слот воркера и просто "спит" внутри процесса. Если у вас 10 сенсоров и 10 слотов — весь кластер встанет в пробку, ожидая файлов.

-

mode="reschedule": Сенсор проверяет условие, и если файла нет — задача завершается, полностью освобождая слот и оперативную память. Планировщик снова поставит её в очередь через poke_interval.

Шаблонизация (Jinja Templating) и op_kwargs В S3KeySensor поле bucket_key поддерживает шаблонизацию "из коробки" — Airflow сам подставляет дату вместо {{ ds }}. Но с Python-функцией сложнее. Если мы просто прочитаем глобальную переменную S3_KEY_TEMPLATE внутри функции, мы получим "сырую" строку с фигурными скобками. Именно поэтому мы используем словарь op_kwargs в определении оператора. Airflow сначала "прогоняет" значения этого словаря через движок шаблонов (превращая {{ ds }} в 2023-01-01), и только потом передает их в функцию как готовые аргументы s3_key и hdfs_dest. Это позволяет не возиться с ручным форматированием дат внутри кода.

  Troubleshooting - когда WebHDFS капризничает?

Работа с Hadoop полна сюрпризов, особенно сетевых. Ошибка 1: requests.exceptions.ConnectionError Диагноз: Airflow не видит Namenode. Лечение: Проверьте docker-compose ps. Сервис hadoop должен быть Up. Проверьте, что в Connection Host указан как namenode (имя контейнера), а не localhost. Ошибка 2: HDFS DataNode is not reachable при загрузке файла WebHDFS работает хитро: сначала вы стучитесь в Namenode ("Я хочу загрузить файл"), она говорит "Ок, пиши на Datanode по адресу 172.18.0.5". И вот тут проблема: этот внутренний IP-адрес Docker-контейнера должен быть доступен вашему воркеру. Лечение: В рамках одной Docker-сети (как у нас) это работает само. Если же Airflow снаружи, а Hadoop внутри Docker, начинаются танцы с пробросом портов и подменой хостов. Именно поэтому мы держим всё в одном docker-compose. Ошибка 3: Permission denied Лечение: HDFS имеет свою систему прав (похожую на Linux). Если вы подключаетесь под пользователем airflow, а папка /user/airflow не создана, будет ошибка. В нашем примере мы отключили проверку прав (dfs_permissions_enabled=false в docker-compose), чтобы не усложнять жизнь на старте.  

Помощь Cursor: Генерация кода для Хуков в Apache AirFlow DAG

  Вам не нужно помнить наизусть методы всех хуков. Используйте Cursor для написания шаблонного кода. Сценарий: Вы хотите не просто загрузить файл, а проверить, не является ли он пустым, перед загрузкой. Промпт для Cursor: "Доработай Python-функцию для Airflow. Перед загрузкой файла в HDFS через WebHDFSHook, проверь размер локального файла. Если он меньше 10 байт, пропусти загрузку и выведи предупреждение. Код должен быть снабжен комментариями." Итог: Мы замкнули цепочку. Данные рождаются в Postgres, экспортируются в S3 и архивируются в HDFS. Попутно мы освоили мощнейший инструмент — Сенсоры, которые позволяют строить событийно-ориентированные (event-driven) пайплайны. Теперь, когда данные лежат в HDFS, настало время их обработать по-взрослому. В следующей статье мы подключим "тяжелую артиллерию" - Apache Spark. Мы разберем SparkSubmitOperator, научимся запускать задачи как на локальном Spark в Docker, так и (в теории) на внешнем кластере. Готовы зажигать со Spark?  

Использованные референсы и материалы