Сегодня рассмотрим, какие ошибки, связанные с DAG, отображаются в пользовательском интерфейсе Apache AirFlow и как дата-инженеру их исправить. А также рассмотрим еще несколько рекомендаций по повышению эффективности этого фреймворка.
4 ошибки с DAG в интерфейсе Apache AirFlow и как их исправить
Сегодня все больше компаний, независимо от их домена и масштаб, стремятся принимать управленческие решения, основанные на данных, практикуя data-driven менеджмент. Поэтому конвейеры сбора и обработки данных должны быть масштабируемыми и надежными. Для создания и эксплуатации таких конвейеров существуют специализированные платформы. Одной из них является Apache AirFlow, фреймворк с открытым исходным кодом для создания, планирования и мониторинга рабочих процессов с помощью Python. AirFlow также предоставляет пользовательский интерфейс для визуализации конвейеров работы с данными, отслеживания их хода и устранения неполадок.
Напомним, AirFlow основан на концепции направленных ациклических графов (DAG, Directed Acyclic Graph), которые состоят из задач. Каждая задача в DAG — это единица работы, например извлечение данных из источника, их преобразование и загрузка в место назначения. Цепочка задач образует конвейер обработки данных, а зависимости между ними могут определять порядок их выполнения. Хотя в целом уровень зрелости Apache AirFlow довольно высок, даже опытные дата-инженеры могут столкнуться с проблемами этой платформы.
Например, DAG не отображается в пользовательском интерфейсе. При наличии ошибки кода DAG будет отображаться во вкладке Broken DAG, выделенном красным на верхней панели GUI. Просмотрев ее, можно понять точную формулировку проблемы, которая исчезнет после ее исправления.
Другой причиной отсутствия отображения DAG в GUI AirFlow может быть отсутствие папки с определением DAG в каталоге DAGS_FOLDER, указанном в файле конфигурации фреймворка. Любой файл Python, который определяет один или несколько объектов DAG в этом каталоге, автоматически анализируется AirFlow и доступен для выполнения.
Другая проблема, связанная с DAG, которую также можно увидеть в GUI Apache AirFlow, вызвана тайм-аутом импорта. Ошибка отображается в разделе Broken DAG с сообщением [/path_to_dagfile/<dag_file_name>.py] Timeout. По умолчанию значение конфигурации dag_file_processor_timeout равно 50. Если увеличить это значение, ошибка исправится.
Еще одной проблемой с DAG может стать непоследовательное появление новых изменений. Это ошибка кода, исправить которую поможет корректное и согласованное именование задач в DAG. Этот параметр, идентифицирующий DAG, является уникальным для каждого ациклического графа. Например, 2 сотрудника работают над одной и той же DAG в среде разработки, используя один и тот же DAG_ID. Первый сотрудник добавил задачу под названием test1, а его коллега добавил задачу под названием test2. Поскольку оба используют один и тот же идентификатор DAG, он передается в среду после внесения изменений обоими членами команды. Чтобы последовательно применить изменения от разных источников к одному и тому же DAG, следует использовать соглашение об именовании задач. Кроме того, для тестирования изменений желательно создавать новый конвейер, а не экспериментировать на работающем. Например, 1-ый сотрудник создает новый DAG и называет его test_dag_new_func_task1, а второй — test_dag_new_func_task2. После этого в пользовательском интерфейсе AirFlow появятся 2 независимых DAG, и оба участника смогут продолжать тестирование, не мешая друг другу.
В заключение отметим проблему отсутствия лог-файла, которая выдается с сообщением о том, что задача не была выполнена или исполняющий ее рабочий процесс завершился ненормально.
Такая проблема может случиться в AirFlow из-за ошибки синтаксического анализа DAG. Хотя планировщик AirFlow и процессор DAG могут анализировать и планировать выполнение задачи, рабочий процесс не может ее выполнить, поскольку в файле DAG есть ошибки кода. В результате задача помечается как неудачная, но при ее выполнении нет ошибок, поскольку фактически она не выполнялась ни разу. В этом случае рекомендуется проверить рабочие журналы Cloud Logging для AirFlow на наличие ошибок синтаксического анализа.
Впрочем, проблема отсутствия лог-файла может также возникнуть из-за чрезмерного использования ресурсов: выполнение задачи AirFlow внезапно прерывается, что приводит к отсутствию журналов. При развертывании AirFlow в Kubernetes также может случиться вытеснение пода, если его ресурсы используются чрезмерно по сравнению с ожидаемым потреблением или заданными настройками. Тогда все экземпляры задач, которые выполняются в этом поде, прерываются, и задача помечается как потерпевшая сбой. В этом случае следует перенастроить под, увеличив ресурсы и снова запустить рабочий процесс AirFlow.
7 рекомендаций по повышению эффективности фреймворка
Чтобы улучшить работу Apache AirFlow с DAG, дата-инженер может использовать следующие рекомендации и лучшие практики:
- Управление зависимостями. Зависимости между задачами могут быть сложными, особенно при работе с большими конвейерами. Рекомендуется использовать объект DAG для определения рабочих процессов и задач с четкими зависимостями. Следует сразу определять рабочие процессы как Python-код. Это упрощает управление версиями и совместную работу с другими членами команды.
- Операторы для определения задач. AirFlow предоставляет широкий спектр операторов для типовых задач, например, PythonOperator для выполнения Python-кода, BashOperator для выполнения команд shell-оболочки и SQLOperator для выполнения SQL-запросов. Подробнее о самых популярных операторах AirFlow мы писали в этом материале.
- Датчики (сенсоры) для ожидания внешних событий. AirFlow предоставляет специальных тип операторов, называемые датчики или сенсоры, которые бездействуют в ожидании внешних событий, таких как запись файлов в каталог или отправка сообщений в очередь. О сенсорах AirFlow мы писали здесь и здесь.
- Подключения (соединения) для хранения учетных данных, таких как пароли базы данных или ключи API. Это позволяет хранить конфиденциальную информацию вне кода и удобнее управлять ею.
- Переменные для хранения значений конфигурации, таких как количество повторных попыток выполнения задачи. Так можно быстро менять настройки без изменения кода.
- Обработка ошибок. Когда задачи терпят сбой, не всегда сразу получается найти основную причину. Для обработки ошибок пригодится функция логирования и повторной попытки для автоматического повторения невыполненных задач. Журналирование позволяет видеть результаты задач, а также любые сообщения об ошибках, которые они генерируют.
- Масштабирование конвейеров. По мере роста объемов данных конвейеры могут становиться медленнее и сложнее в управлении. Чтобы масштабировать конвейеры, целесообразно использовать возможности фреймворков распределенных вычислений, таких как Apache Spark или Flink, а также контейнерную виртуализацию с Kubernetes или Docker.
Узнайте больше про использование Apache AirFlow для дата-инженерии и аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
