Автоматизация рабочих процессов инженерных данных с помощью интеграции Spark и Airflow

Введение: необходимость автоматизированных рабочих процессов данных в инженерии

Инженерные команды сегодня сталкиваются с беспрецедентным потоком данных от датчиков, симуляций, устройств IoT и операционных систем. Обработка этих данных вручную больше невозможна - она вводит задержки, ошибки и узкие места, которые замедляют инновации. Чтобы оставаться конкурентоспособными, организации должны автоматизировать свои конвейеры данных, и два инструмента появились в качестве основы современной инженерии данных: Apache Spark и Apache Airflow. При интеграции они образуют мощную комбинацию, которая обрабатывает все, от потоковой передачи в реальном времени до пакетной обработки, планирования и мониторинга. В этой статье исследуется, как Spark и Airflow работают вместе для автоматизации рабочих процессов инженерных данных, преимущества, которые вы можете ожидать, стратегии внедрения, лучшие практики и примеры из реального мира.

Посмотреть Apache Spark

Apache Spark - это унифицированный аналитический движок с открытым исходным кодом, предназначенный для крупномасштабной обработки данных. В отличие от традиционного MapReduce, Spark сохраняет данные в памяти, делая их в 100 раз быстрее для определенных рабочих нагрузок. Он поддерживает несколько языков (Python, Scala, Java, R) и предоставляет библиотеки для SQL, потоковой передачи, машинного обучения и обработки графов. Для инженерных команд Spark идеально подходит для выполнения сложных преобразований на массивных наборах данных - таких как анализ журналов датчиков, моделирование или агрегирование данных временных рядов.

Основные особенности Spark для инженерных рабочих нагрузок

• Обработка в памяти: Уменьшает I/O диска, ускоряя итеративные алгоритмы и интерактивные запросы.
• Устойчивые распределенные наборы данных (RDD): Недопустимые коллекции, которые могут быть восстановлены, если раздел потерян.
• Spark SQL: Позволяет запрашивать структурированные данные с использованием SQL или DataFrames, которые инженеры могут использовать для специального анализа.
• Проводка: Обеспечивает обработку данных в режиме реального времени для непрерывных источников данных, таких как краевые датчики или производственные линии.
• MLlib: Масштабируемая библиотека машинного обучения для прогнозного обслуживания, обнаружения аномалий и оптимизации.

Кластеры Spark могут быть развернуты на месте или в облаке (AWS EMR, Azure HDInsight, Databricks). Инженеры обычно пишут задания Spark как автономные приложения, которые подаются в кластер через или через API.

Посмотреть Apache Airflow

Apache Airflow — это платформа оркестровки рабочего процесса с открытым исходным кодом. Она позволяет инженерам определять рабочие процессы как направленные ациклические графы (DAG) с использованием кода Python. Каждый узел в DAG представляет собой задачу, а края определяют зависимости. Airflow обрабатывает планирование, повторные записи, мониторинг и оповещение, что делает его инструментом для автоматизации сложных конвейеров данных. В отличие от рабочих мест cron или простых скриптов, Airflow предоставляет богатый пользовательский интерфейс для визуализации состояния задачи, журналов и истории выполнения.

Основные концепции в потоке воздуха

• DAG (Прямой ациклический график): Сбор задач с определёнными зависимостями. Никакие циклы не допускаются, обеспечивая детерминированное выполнение.
• Операторы: Шаблоны для отдельных задач. Примеры включают , и .
• Сенсоры: Специальные задачи, которые ждут внешних событий (например, поступление файла, ответ API).
• XComs: Механизм перекрестной связи для передачи небольших объемов данных между задачами.
• Бассейны и усилители; Исполнители:Управление параллельным выполнением задач и распределением ресурсов.

Airflow может быть развернут на одном сервере, в кластере Kubernetes или с использованием управляемых сервисов, таких как Google Cloud Composer или Amazon Managed Workflows для Apache Airflow (MWAA).

Преимущества интеграции Spark и Airflow

Когда Spark и Airflow объединяются, они охватывают весь жизненный цикл конвейера данных - от приема данных до преобразования, загрузки и мониторинга.

Automation & Оркестрация

Airflow автоматизирует отправку, мониторинг и повторную запись заданий Spark. Вместо ручного выполнения команд или планирования их с помощью cron инженеры определяют DAG, который запускает приложения Spark на кластере. Это устраняет человеческую ошибку и гарантирует, что данные обрабатываются последовательно, даже во время праздников или в нерабочее время.

Масштабируемость и усилие; Управление ресурсами

Spark обрабатывает тяжелую работу распределенных вычислений, масштабируя горизонтально для обработки терабайт данных. Airflow дополняет это, управляя общим рабочим процессом, гарантируя, что зависимые задачи (например, проверки качества данных, загрузка) выполняются только после того, как рабочие места Spark увенчаются успехом. Airflow также может интегрироваться с менеджерами кластеров (YARN, Kubernetes) для динамического распределения ресурсов для каждой задачи Spark.

Надежность и усилие; наблюдаемость

Airflow обеспечивает встроенные повторные записи, оповещения по электронной почте и графический вид выполнения. Если работа Spark не удаётся из-за временной ошибки (например, дефицита кластерных ресурсов), Airflow может перезапустить его с обратной связью. Инженеры могут проверять журналы непосредственно из интерфейса Airflow, сокращая время отладки. Эта надежность имеет решающее значение для инженерных конвейеров данных, которые питают приборные панели, отчеты или модели машинного обучения.

Гибкость и усилие; кастомизация

Комбинация позволяет инженерам проектировать сложные рабочие процессы, которые включают в себя не только задачи Spark, но и извлечение данных (например, из API или баз данных), валидацию и этапы уведомлений. DAG на основе Python Airflow могут включать любую логику, в то время как библиотеки обработки Spark обрабатывают преобразования, специфичные для домена. Эта гибкость означает, что один и тот же конвейер может адаптироваться к новым источникам данных или бизнес-правилам без переписывания уровня оркестровки.

Реализация интеграции

Создание Spark и Airflow вместе требует тщательного планирования инфраструктуры, структуры кода и операций. Ниже приведен пошаговый подход.

Шаг 1: Подготовьте инфраструктуру

Для разработки можно использовать одноузловой экземпляр Spark (локальный режим) и локальную установку Airflow. Для производства рассмотрим облачные сервисы: Databricks для Spark и Cloud Composer или MWAA для Airflow. Обеспечить сетевое подключение между Airflow и Spark — обычно Airflow отправляет задания через REST API или через по SSH.

Шаг 2: Установите необходимые поставщики воздушного потока

Airflow использует пакеты провайдеров для взаимодействия с внешними системами. Для Spark установите пакет . К ним относятся такие операторы, как и . Если вы используете Databricks, установите .

pip install apache-airflow-providers-apache-spark

Шаг 3: Настройка соединений

В интерфейсе Airflow перейдите в Admin > Connections и добавьте соединение Spark. Вам нужно будет указать основной URL (например, или ), режим развертывания и любую необходимую аутентификацию. Для Databricks предоставьте URL рабочего пространства и персональный токен доступа.

Шаг 4: Напишите код приложения Spark

Развивайте свою работу Spark как скрипт Python (или Scala / Java JAR), который считывает исходные инженерные данные, применяет преобразования и записывает результаты в целевую систему (например, файлы Parquet в S3, базу данных).

Шаг 5: Определите Airflow

Создайте DAG, который запланирует и организует работу Spark. Ниже приведен упрощенный пример с использованием :

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime

default_args = {
 'owner': 'engineering',
 'depends_on_past': False,
 'retries': 2,
 'retry_delay': timedelta(minutes=5),
}

with DAG(
 dag_id='engineering_data_pipeline',
 start_date=datetime(2024, 1, 1),
 schedule_interval='0 2 * * *', # daily at 2 AM
 catchup=False,
 default_args=default_args,
) as dag:

 extract_sensor_data = BashOperator(
 task_id='extract_sensor_data',
 bash_command='python /path/to/extract.py',
 )

 transform_sensor_data = SparkSubmitOperator(
 task_id='transform_sensor_data',
 application='/path/to/spark_job.py',
 conn_id='spark_default',
 conf={'spark.executor.memory': '4g'},
 java_class=None,
 )

 load_to_warehouse = BashOperator(
 task_id='load_to_warehouse',
 bash_command='python /path/to/load.py',
 )

 send_notification = EmailOperator(
 task_id='send_notification',
 to='[email protected]',
 subject='Pipeline Complete',
 html_content='<h3>Engineering data pipeline finished successfully.</h3>',
 )

 extract_sensor_data >> transform_sensor_data >> load_to_warehouse >> send_notification

Шаг 6: Испытание и развертывание

Запустите DAG вручную в Airflow для проверки каждого шага. Мониторинг журналов вакансий Spark через интерфейс Airflow или сервер истории Spark. После проверки установите активную DAG и позвольте ей работать по расписанию.

Лучшие практики для трубопроводов Spark + Airflow

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

Распределение ресурсов и усилитель; настройка

• Сопоставьте исполнителей Spark с емкостью кластера: Используйте параметр Airflow для установки , и на основе размера кластера. Переоборудование может вызвать спор о ресурсах; недооборудование замедляет работу.
• Динамическое распределение рычагов: Позволяет , чтобы позволить исполнителям масштаба Spark вверх/вниз на основе рабочей нагрузки.
• Использовать пулы ресурсов в потоке воздуха: Для сред с несколькими DAG, определить пулы, чтобы ограничить количество одновременных задач Spark и предотвратить перегрузку кластера.

Обработка ошибок & Retries

• Установите повторные попытки уровня DAG: Используйте и для автоматического повторного повторения неудавшихся задач. Для переходных ошибок Spark (например, потерянный исполнитель) это позволяет избежать ручного вмешательства.
• Внедряйте пользовательские датчики: Если ваш конвейер зависит от внешних данных, используйте датчик (например, ) вместо фиксированного графика.
• Добавить контрольно-пропускной пункт в Spark: Для длительных работ периодически сохраняйте промежуточные результаты.Если задача не срабатывает и повторяется, Spark может возобновить работу с последнего контрольно-пропускного пункта, а не перерабатывать все данные.

Мониторинг и маркировка; оповещение

• Включить оповещение Airflow: Настройте уведомления по электронной почте или Slack для сбоев в работе и промахов SLA.
• Агрегация журналов: Журналы Ship Spark (водитель и исполнитель) в централизованную систему, такую как Elasticsearch или CloudWatch.
• Монитор кластерных метрик Spark: Используйте встроенную систему метрик Ganglia, Prometheus или Spark. Предупреждение о высокой разливе перетасовок, длительных временах GC или сбоях в работе.

Code Structure & - версия

• Сохраняйте DAGs наклонными: Избегайте больших вычислений в задачах Airflow. Используйте Spark для обработки; Airflow должен только оркестровать.
• Используйте DAG-версию: Храните файлы DAG в репозитории Git и развертывайте через CI/CD. Пометьте каждую версию DAG, чтобы соответствовать версии кода Spark.
• Параметризуйте среды: Используйте переменные Airflow или переменные среды для настройки путей файлов, соединений с базой данных и конечных точек кластера — никогда не кодируйте их жестко.