Airflow — идеальный выбор для конвейеров данных, то есть для оркестрации и планирования ETL. Он широко применяется и популярен для создания конвейеров передачи данных будущего. Он обеспечивает обратное заполнение, управление версиями и происхождение с помощью функциональной абстракции.
Функциональное программирование — это будущее.
Оператор определяет единицу в рабочем процессе, DAG — это набор Задач. Операторы обычно работают независимо, на самом деле они могут работать на совершенно двух разных машинах. Если вы инженер данных и работали с Apache Spark или Apache Drill, вы, вероятно, знаете, что такое DAG! Такая же концепция и у Airflow.
Создавайте конвейеры данных, которые:
Идемпотентны.
Детерминированы.
Не имеют побочных эффектов.
Используют неизменные источники и направления.
Не обновляются, не добавляются.
Модульность и масштабируемость — основная цель функциональных конвейеров данных.
Если вы работали с функциональным программированием с помощью Haskell, Scala, Erlang или Kotlin, вы удивитесь, что это то, что мы делаем в функциональном программировании, и вышеперечисленные пункты относятся к функциональному программированию, да! Вы правы. Функциональное программирование — это мощный инструмент будущего.
Если у вас есть ETL / Data Lake / Streaming Infrastructure как часть платформы разработки данных, у вас должен быть кластер Hadoop / Spark с некоторым дистрибутивом, таким как Hortonworks, MapR, Cloudera и т. д. Поэтому я собираюсь рассказать о том, как вы можете использовать ту же инфраструктуру, где у вас есть Apache Hadoop / Apache Spark Cluster, и вы используете его для создания Airflow Cluster и его масштабирования.
Если у вас много заданий ETL и вы хотите организовать и составить расписание с помощью некоторых инструментов планирования, у вас есть несколько вариантов, например Oozie, Luigi и Airflow. Oozie основан на XML, и мы взяли его 2019 году! :), Luigi чуть не выбросили после того, как Airflow родился на Airbnb.
Почему не используем Luigi с Airflow?
У Airflow есть собственный планировщик, в котором Luigi требует синхронизировать задачи в задании cron.
С Luigi навигация по пользовательскому интерфейсу становится сложной задачей.
В Luigi сложно создавать задачи.
Luigi не масштабируется из-за тесной связи с заданиями Cron.
Повторный запуск процесса в Luigi невозможен.
Luigi не поддерживает распределенное выполнение, так как оно плохо масштабируется.
До появления Airflow я использовал Luigi для поддержки рабочего процесса моделей машинного обучения с помощью Scikit-learn, Numpy, Pandas, Theano и т. д.
В последнем сообщении блога мы обсудили, как настроить мультинодовый кластер Airflow с Celery и RabbitMQ без использования Ambari.
Ага, переходим к главному.
Как настроить мультинодовый кластер Airflow в Hadoop Spark Cluster, чтобы Airflow мог запускать задания Spark/Hive/Hadoop Map Reduce, а также выполнять координацию и планирование.
Давайте сделаем это!
Вы должны использовать airflow-ambari-mpack (пакет управления Apache Airflow для Apache Ambari), я использовал реализацию с открытым исходным кодом от FOSS Contributor https://github.com/miho120/ambari-airflow-mpack, Спасибо за ваш вклад.
Следующие шаги:
Из предыдущего сообщения в блоге вы должны выполнить шаги с 1 по 4, чтобы установить RabbitMQ и другие пакеты.
Ref. Как настроить мультинодовый кластер Airflow с помощью Celery и RabbitMQ
- Устанавливаем Apache MPack для Airflow
a. git clone https://github.com/miho120/ambari-mpack.git
b. stop ambari server
c. install the apache mpack for airflow on ambari server
d. start ambari server- Добавляем Airflow Service в Ambari
После успешного выполнения вышеуказанных шагов вы можете открыть интерфейс Ambari
http://<HOST_NAME>:8080Откройте пользовательский интерфейс Ambari (Ambari UI), нажмите Действия -> Добавить службу. (Actions -> Add Service)
HDP Ambari Dashboard
Если шаг 1 выполнен успешно, вы сможете увидеть Airflow как часть службы Ambari.
Сервис Airflow в Ambari
Вы должны выбрать, на каком узле вы хотите установить веб-сервер, планировщик и воркер. Я бы порекомендовал установить Airflow на веб-сервер, планировщик на master ноде, то есть на узле имени, и на Install Worker на data-нодах.
Ambari Master нода / Name нода для Airflow
Как вы можете видеть на изображении выше, веб-сервер Airflow и планировщик Airflow установлены на Name ноде кластера Hadoop / Spark.
Как вы можете видеть на скриншоте выше, служба Airflow Worker установлена на Data ноде кластера.
В итоге, у меня есть 3 воркер (worker) ноды на трех data нодах.
Сервис Airflow в Ambari
Ambari UI: 3 воркера в Airflow
Вы можете добавлять рабочие нода столько, сколько хотите, вы можете добавлять / удалять воркеров, когда захотите. Эта стратегия может масштабироваться по горизонтали + вертикали.
Конфигурация Airflow в Ambari:
Нажмите на Airflow Service, а затем на Config в пользовательском интерфейсе Ambari.
Конфигурация Airflow в Ambari
- Смените Executor
executor = CeleryExecutor
В разделе Advanced airflow-core-site укажите Executor как CeleryExecutor.
- SQL Alchemy Connection
sql_alchemy_conn = postgresql+psycopg2://airflow:airflow@{HOSTNAME}/airflow
SQL Alchemy Connection
Измените соединение SQL Alchemy на соединение postgresql, пример приведен выше.
- URL-адрес брокера
broker_url= pyamqp://guest:guest@{RabbitMQ-HOSTNAME}:5672/
celery_result_backend = db+postgresql://airflow:airflow@{HOSTNAME}/airflow
URL-адрес брокера и Celery result backend для Airflow
- Прочие
dags_are_paused_at_creation = True
load_examples = False
Конфигурация Airflow-core-site.
После того, как все эти изменения будут внесены в конфигурацию Ambari Airflow, Ambari попросит вас перезапустить все затронутые службы, перезапустите службы и нажмите Service Actions -> InitDB.
Airflow Initdb из Ambari
А затем запустите службу airflow. Теперь у вас должен получиться мультинодовый кластер Airflow.
Кое-что из Чек-листа для проверки служб для мультинодового кластера Airflow:
- Очереди RabbitMQ должны быть запущены:
- Подключения RabbitMQ должны быть активными:
- Каналы RabbitMQ должны быть запущены:
- Отображение Celery Flower
Celery Flower — это веб-инструмент для мониторинга и управления кластерами Celery. Номер порта по умолчанию — 5555.
Вы также можете видеть здесь, что 3 рабочих находятся в сети, и вы можете отслеживать одну единицу «задачи» Celery здесь.
Подробнее о Celery Flower: https://flower.readthedocs.io/en/latest/
Обратите внимание, что вы также можете запустить «Celery Flower», веб-интерфейс, созданный поверх Celery, для наблюдения за своими рабочими. Вы можете использовать команду быстрого доступа airflow flower, чтобы запустить веб-сервер Flower.
nohup airflow flower >> /var/local/airflow/logs/flower.logs &Мы закончили установку и настройку мультинодовый кластер Airflow на Ambari HDP Hadoop / Spark Cluster.
Я столкнулся с некоторыми проблемами, о которых я расскажу в следующей статье в блоге.
