Продолжая тему тестирования DAG в Apache Airflow, сегодня рассмотрим следующий этап проверки качества ПО – разработку интеграционных тестов. Разберемся, как при этом дата-инженер может использовать Docker Compose и Pytest, а также познакомимся с возможностями REST API самого популярного в Big Data batch-оркестратора.
Идеи и инструменты интеграционного тестирования DAG в Apache AirFlow
Apache Airflow не случайно считается самым популярным средством планирования и оркестрации рабочих процессов в области Big Data. Он часто используется в конвейерах обработки данных для организации задач, которые могут взаимодействовать с внешними зависимостями. Эти внешние зависимости обычно имитируются при написании модульных тестов для цепочек задач в виде направленных ациклических графов (DAG). Однако, при разработке ПО важно убедиться не только в корректной работе отдельных единиц программного кода, т.е. модулей. На следующем уровне пирамиды тестирования лежат интеграционные тесты, когда отдельные программные модули объединяются и тестируются как группа. Таким образом, интеграционный тест для DAG Airflow должен проверить, что рабочий процесс, смоделированный DAG для конвейера данных, работает должным образом, особенно при взаимодействии с несколькими внешними зависимостями.
Рассмотрим некоторые аспекты интеграционного тестирования DAG AirFlow на следующем примере. Предположим, интеграционный тест для DAG, имеет только 2 задачи: получить документ из хранилища документов на основе заданного идентификатора и установить для обрабатываемого поля значение true. В интеграционном тесте нужно будет запустить веб-службу Airflow со всеми ее зависимостями, службу хранилища документов и процесс, который будет запускать тест. Здесь понадобится Docker Compose — инструмент для определения и управления многоконтейнерными приложениями Docker, открытой платформа, которая позволяет разработчикам упаковывать приложения со всеми необходимыми зависимостями в контейнеры.
Для рассматриваемого примера следует определить YAML-файл Docker Compose, который будет запускать необходимые службы:
- для веб-сервиса Airflow, планировщика и рабочего сервиса используется готовый образ от Bitnami, где уже установлен фреймворк, что упрощает настройку кластера;
- служба PostgreSQL используется для управления состоянием;
- служба Redis нужна для планирования задач.
- служба MongoDB выполняет роль хранилища документов, которое является внешней зависимостью рассматриваемого в тестируемом DAG;
- служба запуска тестов будет запускать интеграционные тесты с помощью Pytest – тестовой Python-среды, которая упрощает настройку и удаление тестов. Она построена с использованием образа планировщика AirFlow от Bitnami, поскольку для тестов требуется пакет A Установить дополнительные пакетов в образ от Bitnami можно через настройку файла /bitnami/python/requirements.txt.
Кроме того, поскольку исполнителю тестов требуется веб-служба Airflow для своего REST API, перед запуском теста службе выполнения тестов необходимо дождаться готовности веб-службы AirFlow. Поскольку в тестовой среде требуется конфиденциальная информация, которая используется для настройки учетных данных для Airflow, нужно будет хранить эту информацию вне YAML-файла Docker Compose (docker-compose.yml), который обычно фиксируется в рабочем инструменте управления версиями исходного кода. Рекомендуется сохранить его в файле .envrc и игнорировать этот файл во время коммитов.
Чтобы упростить работу с Pytest, можно использовать Makefile для инкапсуляции команд, используемых для запуска тестов. В частности, можно просто запустить make Integration_test, чтобы запустить интеграционные тесты, и запустить make manual_testing, чтобы запустить программу запуска тестов в режиме ручного тестирования.
Таким образом, с помощью Docker Compose можно настроить тестовую среду для тестируемого DAG Airflow, а также все внешние зависимости, чтобы запускать наборы интеграционных тестов с помощью Pytest или аналогичного инструмента. Как мы уже упомянули, исполнителю тестов требуется веб-служба Airflow, коммуникации с которой происходят посредством REST API. Что представляет собой REST API этого фреймворка, рассмотрим далее.
Краткий ликбез по REST API Apache Airflow
Чтобы упростить работу дата-инженера, Apache Airflow поддерживает ряд конечных точек REST API для своих объектов. Большинство конечных точек принимают JSON в качестве входных данных и возвращают ответы JSON. Термин ресурс относится к одному типу объекта в метаданных Airflow. API разбивается по соответствующему ресурсу его конечной точки. Имя ресурса обычно имеет множественное число и выражается в стиле camelCase, например, dagRuns. Имена ресурсов используются как часть URL-адресов конечных точек, а также в параметрах и ответах API.
Платформа поддерживает операции Create, Read, Update и Delete для большинства ресурсов. Некоторые конечные точки имеют особое поведение в виде исключений. Как и в любом RESTful API, чтобы создать ресурс, на конечную точку отправляется HTTP-запрос POST с необходимыми метаданными ресурса в теле запроса. Ответ возвращает код ответа 201 Created при успешном использовании метаданных ресурса, включая его внутренний идентификатор, в теле ответа.
Для чтения ресурса или для вывода списка ресурсов используется HTTP-запрос GET с идентификатор ресурса в параметрах запроса для чтения определенных данных. Ответ обычно возвращает код ответа 200 OK в случае успеха с метаданными ресурса в теле ответа. Если запрос GET не включает конкретный идентификатор ресурса, он рассматривается как запрос списка. Ответ обычно возвращает код ответа 200 OK в случае успеха с объектом, содержащим список метаданных ресурсов в теле ответа.
Для обновления ресурса нужен его идентификатор, обычно отправляемый с помощью HTTP-запроса PATCH с полями, которые нужно изменить в теле запроса. Ответ возвращает код ответа 200 OK в случае успеха с информацией об измененном ресурсе в теле ответа. Удаление ресурса требует его идентификатора и выполняется через HTTP-запрос DELETE. Ответ возвращает код ответа 204 No Content в случае успеха.
Маска обновления доступна в качестве параметра запроса в конечных точках исправления. Он используется для уведомления API, какие поля нужно обновить. Использование update_mask упрощает обновление объектов, указывая серверу, какие конкретные поля в объекте следует обновить вместо тотального обновления всех полей. Запрос на обновление игнорирует любые поля, не указанные в маске поля, оставляя их текущими значениями. Например, следующий участок кода считывает ресурс /resource/my-id и заменяет значение поля my_field на new-value:
resource = request.get('/resource/my-id').json()
resource['my_field'] = 'new-value'
request.patch('/resource/my-id?update_mask=my_field', data=json.dumps(resource))
Наконец, когда речь идет об интеграционном взаимодействии систем, нельзя забыть про вопросы аутентификации. Apache Airflow поддерживает несколько методов аутентификации (Basic, Kerberos), к которым можно добавить свой собственный, настроив это в конфигурационном файле airflow.cfg. Если нужно проверить, какой сервер аутентификации установлен в данный момент, используется команда airflow config get-value API auth_backends.
В заключение отметим, что для отправки HTTP-запросов к REST API AirFlow подойдет любой HTTP-клиент для тестирования API: Postman, SoapUI, JMeter и пр. Подробнее об этом читайте в нашей новой статье.
Код курса
ADH-AIR
Ближайшая дата курса
Продолжительность
ак.часов
Стоимость обучения
0 руб.
Узнайте больше подробностей про администрирование и эксплуатацию Apache AirFlow для эффективной организации ETL/ELT-процессов в аналитике больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
