Как писать на Go для AirFlow: новый SDK в релизе 3.0

Зачем в Apache AirFlow 3.0 добавлена поддержка Go и как работает этот экспериментальный SDK: возможности и ограничения разработки и запуска задач на компилируемом языке программирования.

Мультиязычность в Apache AirFlow 3.0

Одной из ключевых новинок недавно выпущенного Apache AirFlow 3.0, о котором мы писали здесь, стала его мультиязычность. Теперь фреймворк поддерживает не только Python, но и Typescript, а также Java, Scala и Go. Разумеется, основным языком разработки в ETL-оркестраторе остается Python, однако теперь Airflow позволяет пользователям определять задачи DAG и на других языках. В частности, в релизе 3.0 доступен SDK для Go. Пока эта функция находится в статусе экспериментальной. Однако, возможно уже в ближайших выпусках она станет более устойчивой и рекомендуемой для использования в производстве.

Вообще мультиязычность, т.е. поддержка нескольких языков, полезна для переноса рабочих процессов из других ETL-инструментов в Airflow без долгого и дорогого рефакторинга. Как это работает, разберем дальше на примере экспериментального Go SDK, а пока отметим, что еще запускать задачи на других языках в Airflow можно следующим образом:

использовать BashOperator для запуска скрипта на другом языке, например, написанных на JavaScript или R. Например, одна задача DAG с использованием Bash-оператора запускает JavaScript для запроса внешнего API и передает полученный ответ в XCom, чтобы вторая задача могла обработать эти данные в скрипте, написанном на R. Разумеется, для этого необходимо установить языковые пакеты JavaScript и R на уровне операционной системы. Далее следует написать файлы скриптов на JavaScript и R, и сделать их доступными для среды Airflow. Выполнить эти скрипты из DAG будет задача с BashOperator – оператора, который выполняет команды или скрипты командной оболочки bash из Airflow.
с помощью KubernetesPodOperator для запуска любого образа Docker, который может включать код на любом языке. Для этого надо создать пользовательский образ Docker, содержащий скрипт и запустить его из KubernetesPodOperator с обязательным аргументом image. KubernetesPodOperator запускает любой допустимый образ Docker, предоставленный ему в выделенном поде Kubernetes в кластере Kubernetes, абстрагируя API-вызовы этой платформы управления контейнерными приложенями, чтобы запускать и выполнять код DAG.

Data Pipeline на Apache Airflow

Код курса

AIRF

Ближайшая дата курса

2 июня, 2025

Продолжительность

24 ак.часов

Стоимость обучения

72 000

Экспериментальный Go SDK в релизе 3.0

В общем случае Go считается компилируемым языком, кроме таких специфических проектов, как интерпретатор YAEGI, который обеспечивает работу исполняемых скриптов и плагинов Go во встроенных интерпретаторах или интерактивных оболочках поверх среды выполнения Go. Поэтому для поддержки этого языка программирования в Python-среде Airflow необходимы средства, которые позволят скомпилировать задачи в двоичный файл и зарегистрировать их внутри рабочего процесса для выполнения.

Для этого Go SDK в Airflow 3.0 использует интерфейс выполнения задач Task API. Однако, этот API не предоставляет средства для получения ExecuteTaskWorkload самого рабочего процесса Go. Поэтому сейчас используется gopher-celery для получения задач от брокера Redis. Это базовый механизм для эффективного создания и использования задач Celery на стороне Go. Celery — это асинхронная очередь задач или заданий с открытым исходным кодом, основанная на распределенной передаче сообщений. Она поддерживает планирование, но больше фокусируется на выполнении операций в режиме реального времени. По умолчанию в Celery по умолчанию не включен бэкенд результатов. Поэтому, чтобы отслеживать задачи или получать возвращаемые значения, Celery должен где-то хранить или отправлять куда-то состояния, откуда их можно извлечь позже. Обычно для этого используется SQLAlchemy, Django ORM, Memcached, RabbitMQ, QPid или Redis. Также можно определить свой собственный бэкенд для хранения и передачи результатов. Чтобы гарантировать освобождение ресурсов, нужно вызвать метод get() или forget() на каждом экземпляре AsyncResult, возвращенном после вызова задачи.

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

В настоящее время Go SDK в Airflow 3.0 может:

получать задачи из очередей Celery;
запускать зарегистрированных задач, но без поддержки DAG-версий или загрузки нескольких различных пакетов;
отправлять heartbeat0сигналы и сообщать о конечном состоянии конечного экземпляра задачи;
разрешать доступ к переменным.

Текущая реализация Go SDK в Airflow 3.0 пока не поддерживает чтение подключений и перевод задач в состояние, кроме успешного или неудачного/готового к повторной попытке: отложенное, неудачное без повторных попыток и пр. Также отсутствует HTTP-сервер журналов для просмотра логов текущих задач, сами логи удаленных задач на S3/GCS и пр. Кроме того, пока нельзя читать и записывать XCom-объекты с сервера API и из других бэкэндов XCom. Возможно, это будет реализовано в следующих итерациях.

Еще в планах разработчиков стабилизировать Edge Executor API и написать для него Go-клиент. Планируется добавить поддержку нескольких версий путем компиляции задач/пакетов в плагины и использования Go-плагина, подобно тому, как работают провайдеры Terraform. Это позволит использовать код исполнителя и код задачи в отдельных процессах. Также расширятся возможности исполнителя: один исполнитель сможет выполнять различные пакеты и разные версии задач.

Таким образом, с появлением нового SDK для других языков программирования Apache AirFlow увеличивает количество своих вариантов использования в инженерии данных.