В этой статье для обучения дата-инженеров и администраторов кластера Apache AirFlow рассмотрим, как обновить этот ETL-планировщик, используя концепцию сине-зеленого развертывания. Также рассмотрим, с какими ошибками можно столкнуться, выполняя миграцию базы данных метаданных и как их решить.
Сине-зеленое развертывание для обновления AirFlow
Как и любое программное обеспечение, Apache AirFlow нужно периодически обновлять, поскольку в каждом новом выпуске устраняются ошибки и появляются новые функции, которые упрощают работу дата-инженера по управлению пакетными конвейерами обработки больших данных. Более новые версии AirFlow могут содержать миграции базы данных, поэтому необходимо запустить обновление базы данных с помощью изменений схемы. Это безопасно, даже если нет необходимости выполнять миграции.
Перед любой миграцией рекомендуется сделать резервную копию базы данных метаданных. Чтобы это резервное копирование было согласованным, его следует делать после закрытия всех экземпляров AirFlow. Если резервная копия не была сделана и миграция не удалась, можно восстановить базу данных из резервной копии и повторить миграцию. Сбой может возникнуть из-за нарушения сетевого соединения между CLI-интерфейсом и базой данных во время миграции, поэтому создание резервной копии позволит избежать многих проблем.
Если есть настраиваемое развертывание на основе virtualenv или Docker-контейнеров, нужно запускать обновление базы данных вручную. Но иногда обновление происходит автоматически, например, если это действие встроено в развертывание после установки. Например, при использовании Helm-диаграммы для Apache AirFlow с включенными перехватчиками после обновления, обновление базы данных происходит автоматически сразу после установки нового ПО. Аналогично все облачные решения AirFlow автоматически выполняют обновление.
Если нужно запустить сценарий обновления в автономном режиме, можно использовать флаг –r или –revision-range, чтобы получить операторы SQL, которые будут выполняться. Эта функция поддерживается в PostgreSQL и MySQL, начиная с версии AirFlow 2.0.0 и в MSSQL, начиная с версии AirFlow 2.2.0. Чтобы вручную обновить базу данных, следует запустить команду обновления базы данных AirFlow в виртуальной среде, либо в контейнерах, которые предоставляют доступ к CLI-интерфейсу.
Часто при обновлении AirFlow возникает вопрос: какие компоненты следует обновить прежде всего? Помимо пользовательских DAG, также есть YAML-файлы свойств об их конфигурации, подключения и переменные. Хотя подключения и переменные останутся прежними, необходимо внести изменения в DAG и файлы свойств, чтобы они соответствовали новым соглашениям. Современные DevOps-инженеры часто используют стратегию сине-зеленого развертывания для обновления ПО, которая сокращает время простоя за счет запуска двух идентичных производственных сред: синей и зеленой. В любой момент активна только одна из этих сред: она обслуживает весь трафик, а старое задание заменяется новым, когда оба запущены и только одно из них обрабатывает данные. При этом немного снижается скорость отправки новых заданий, которая зависит в основном от количества доступных ресурсов. Сине-зеленое развертывание может занять несколько минут, что мы рассматривали здесь.
Для обновления AirFlow согласно стратегии сине-зеленого развертывания следует создать новые ветки разработки и производства, в которых будут внесены изменения в DAG и файлы свойств. Далее эти новые ветки будут развернуты на новом сервере, тогда как старый сервер продолжает работать без перебоев, чтобы выполнить откат назад в случае каких-либо сбоев.
Операторы DAG перемещаются в новые пакеты, поэтому нужно убедиться, что операторы импорта верны, при том, что Xcom и Provide_context изменены. Если изменился стандартный формат даты выполнения изменился с %Y-%m-%d на YYYY-MM-DD, нужно убедиться, что новый формат даты обновлен соответствующим образом. Это обеспечит корректность запуска заданий по запланированному расписанию.
Если при обновлении потребуются ресурсы за пределами разрешений стандартной роли выполнения AirFlow, администратору кластера необходимо предоставить соответствующую учетную запись. Сам процесс обновления ETL-фреймворка не очень сложен, однако, при миграции базы данных, где хранятся метаданные, могут возникнуть ошибки. Какие они могу быть и как с ними справиться, рассмотрим далее.
Обновления базы данных: проблемы и решения
Одной из частых проблем при обновлении базы метаданных связано с неверной кодировкой. Если в качестве хранилища метаданных использовался MySQL, миграция может завершиться неудачно с ошибками типа «слишком большой размер ключа», «отсутствующие индексы» и пр. Для базы данных MySQL 8 рекомендуется набор символов/сопоставление: utf8mb4 и utf8mb4_bin соответственно. MySQL ограничивает размер ключа индекса, а с utf8mb4 размеры ключа индекса AirFlow могут быть слишком большими для обработки этой СУБД. Поэтому рекомендуется для всех ключей «ID» использовать набор символов utf8, что эквивалентно utf8mb3 в MySQL 8. Это ограничивает размер индексов, чтобы MySQL мог их обрабатывать. Это следует сделать до миграции, предварительно сделав резервную копию базы данных, чтобы восстановить ее в случае ошибки.
Чтобы понять, какие из ваших таблиц нуждаются в исправлении, их следует просмотреть, задав следующие SQL-команды:
SHOW CREATE TABLE task_reschedule; SHOW CREATE TABLE xcom; SHOW CREATE TABLE task_fail; SHOW CREATE TABLE rendered_task_instance_fields; SHOW CREATE TABLE task_instance;
В столбцах dag_id, run_id, task_id и key должен быть явно указан набор символов utf8 или utf8mb3. Если поля не имеют кодировки, сопоставление установлено на utf8mb4, или набор символов и сопоставление установлены на utf8mb4, возникнут ошибки. Нужно исправить поля, которые имеют неправильный набор символов/сопоставление.
Также следует отбросить индексы внешнего ключа для таблиц, которые нужно изменить. Их не надо удалять их все, достаточно только для тех таблиц, которые подлежат изменению, например:
ALTER TABLE task_reschedule DROP FOREIGN KEY task_reschedule_ti_fkey; ALTER TABLE xcom DROP FOREIGN KEY xcom_task_instance_fkey; ALTER TABLE task_fail DROP FOREIGN KEY task_fail_ti_fkey; ALTER TABLE rendered_task_instance_fields DROP FOREIGN KEY rtif_ti_fkey;
Также надо изменить поля идентификатора, чтобы они имели правильный набор символов/кодировку:
ALTER TABLE task_instance MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_reschedule MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE rendered_task_instance_fields MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE rendered_task_instance_fields MODIFY dag_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_fail MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_fail MODIFY dag_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE sla_miss MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE sla_miss MODIFY dag_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY dag_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE task_map MODIFY run_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY task_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY dag_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY run_id VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin; ALTER TABLE xcom MODIFY key VARCHAR(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin;
В зависимости от версии AirFlow индексы могут немного отличаться: например, map_index был добавлен в 2.3.0. Сохранив выходные данные SHOW CREATE TABLE, можно найти правильное ограничения имени и использования из вывода SHOW CREATE TABLE:
ALTER TABLE <TABLE> ADD CONSTRAINT `<CONSTRAINT_NAME>` <CONSTRAINT>
Они приведут базу данных в состояние, откуда можно выполнить миграцию на новую версию AirFlow, просто запустив команду обновления базы данных фреймворка. Однако в некоторых случаях миграция может обнаружить некоторые старые, устаревшие и неверные данные и переместить их в отдельную таблицу. В этом случае в пользовательском интерфейсе веб-сервера будут показаны предупреждения о найденных данных. Рекомендуется проверить эти данные, чтобы определить, надо ли их сохранить или удалить. Если данные были повреждены и остались от каких-то ошибок, их можно безопасно удалить, используя разные способы.
При наличии прямого доступа к базе данных с помощью специализированных инструментов, можно удалить такую таблицу, переименовать ее или переместить в другую базу данных. Также можно использовать команду оболочки AirFlow db.
При развертывании AirFlow в Kubernetes удаление можно выполнить в любом из подов AirFlow: веб-сервере или планировщике, используя команду
kubectl exec -it <your-webserver-pod> python
В оболочке Python можно запустить следующие команды, заменив <table> фактическим именем таблицы, напечатанным в предупреждающем сообщении:
from airflow.settings import Session
session = Session()
session.execute("DROP TABLE _airflow_moved__2_2__task_instance")
session.commit()
В зависимости от размера базы данных и фактической миграции обновление AirFlow может занять довольно много времени. Поэтому рекомендуется сперва сделать копию базы данных и выполнить тестовую миграцию, чтобы понять, сколько времени займет этот процесс. Масштабные обновления занимают больше времени, поскольку добавление новых функций иногда требует реструктуризации базы данных.
Код курса
ADH-AIR
Ближайшая дата курса
Продолжительность
ак.часов
Стоимость обучения
0 руб.
Больше полезных приемов администрирования и эксплуатации Apache AirFlow для дата-инженерии и аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
- https://medium.com/@matt_weingarten/upgrading-AirFlow -environments-afbdd2cc66c3
- https://AirFlow .apache.org/docs/apache-AirFlow /stable/installation/upgrading.html
