Разбор выполнения рабочей задачи в Apache DolphinScheduler

Привет, ребята, я Цай Шуньфэн, старший инженер по данным в WhaleOps, а также коммиттер и член PMC сообщества Apache DolphinScheduler. Сегодня я объясню, как работает задача Worker в Apache DolphinScheduler.

Это объяснение будет разделено на три части:

1. Введение в Apache DolphinScheduler
2. Обзор общей конструкции Apache DolphinScheduler
3. Подробный процесс выполнения рабочих задач

Введение в проект

Apache DolphinScheduler — это распределенная, легко расширяемая, визуальная система планирования рабочих процессов с открытым исходным кодом, подходящая для сценариев корпоративного уровня.

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

Основные характеристики

• Легко использовать

• Визуальные операции DAG: пользователи могут перетаскивать компоненты на странице, чтобы организовать их в DAG (направленный ациклический граф).

• Система плагинов: включает плагины задач, плагины источников данных, плагины оповещений, плагины хранилищ, плагины центра реестра, плагины заданий cron и т. д. Пользователи могут легко расширять плагины по мере необходимости в соответствии с требованиями своего бизнеса.

  
  

• Разнообразные сценарии использования

• Статическая конфигурация: включает планирование рабочих процессов, онлайн- и офлайн-операции, управление версиями и функции обратного заполнения.

• Операции во время выполнения: Предоставляет такие функции, как пауза, остановка, возобновление и подстановка параметров.

• Типы зависимостей: поддерживает широкий набор вариантов и стратегий зависимостей, адаптируясь к большему количеству сценариев.

• Передача параметров: поддерживает параметры запуска на уровне рабочего процесса, глобальные параметры, локальные параметры на уровне задач и динамическую передачу параметров.

  
  

• Высокая надежность

• Децентрализованная конструкция: все сервисы не имеют состояния и могут масштабироваться горизонтально для увеличения пропускной способности системы.

• Защита от перегрузки и отказоустойчивость экземпляра:

• Защита от перегрузки: во время работы главный и рабочий отслеживают использование собственного ЦП и памяти, а также объем задачи. В случае перегрузки они приостанавливают текущий рабочий процесс/обработку задачи и возобновляют ее после восстановления.

• Устойчивость к отказам экземпляра: при выходе из строя главного/рабочего узлов центр реестра обнаруживает, что узел службы находится в автономном режиме, и обеспечивает устойчивость к отказам для экземпляров рабочих процессов или задач, обеспечивая максимальную возможность самовосстановления системы.

Общий дизайн

Архитектура проекта

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

Из диаграммы архитектуры видно, что Apache DolphinScheduler состоит из нескольких основных компонентов:

• Компонент API: служба API в первую очередь управляет метаданными, взаимодействует с пользовательским интерфейсом через службу API или вызывает интерфейсы API для создания задач рабочего процесса и различных ресурсов, необходимых для рабочего процесса.

  
  

• Главный компонент: главный компонент — это контроллер экземпляров рабочего процесса, отвечающий за использование команд, преобразование их в экземпляры рабочего процесса, выполнение разделения DAG, отправку задач по порядку и распределение задач между работниками.

  
  

• Компонент Worker: Worker является исполнителем определенных задач. Получив задачи, он обрабатывает их в соответствии с различными типами задач, взаимодействует с мастером и сообщает о статусе задачи. Примечательно, что служба Worker не взаимодействует с базой данных; с базой данных взаимодействуют только API, мастер и службы оповещений.

  
  

• Служба оповещений: Служба оповещений отправляет оповещения через различные плагины оповещений. Эти службы регистрируются в центре реестра, а главный и рабочий периодически сообщают о тактовых импульсах и текущем статусе, чтобы убедиться, что они могут нормально получать задачи.

Процесс взаимодействия мастера и рабочего

Процесс взаимодействия мастера и рабочего выглядит следующим образом:

• Отправка задач: после того, как мастер завершает разделение DAG, он отправляет задачи в базу данных и выбирает соответствующую рабочую группу для распределения задач на основе различных стратегий распределения.

  
  

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

  
  

• Выполнение задачи: Worker обрабатывает задачу, обновляет статус на выполнение и возвращает мастеру. Мастер обновляет статус задачи и информацию о времени начала в базе данных.

  
  

• Завершение задачи: после завершения задачи воркер отправляет мастеру уведомление о событии завершения, а мастер возвращает подтверждение ACK. Если ACK не получен, воркер продолжит повторять попытки, чтобы убедиться, что событие задачи не потеряно.

Прием рабочих заданий

При получении задания работником выполняются следующие операции:

• Заполняет информацию о хосте.
• Генерирует путь журнала на рабочей машине.
• Создает исполнитель рабочей задачи, который отправляется в пул потоков для выполнения.

Рабочий проверяет, не перегружен ли он; если да, то отклоняет задачу. Получив обратную связь о сбое распределения задач, мастер продолжает выбирать другого рабочего для распределения задач на основе стратегии распределения.

Процесс исполнения работником

Конкретный процесс выполнения рабочих задач включает в себя следующие этапы:

1. Инициализация задачи: инициализирует среду и зависимости, необходимые для задачи.
2. Выполнение задачи: выполняет конкретную логику задачи.
3. Завершение задачи: после завершения выполнения задачи отправляет отчет о результатах выполнения задачи главному узлу.

Далее мы подробно рассмотрим конкретный процесс выполнения задачи.

Перед началом выполнения задачи сначала инициализируется контекст. На этом этапе устанавливается время начала задачи. Для обеспечения точности выполнения задачи необходимо синхронизировать время между мастером и работником, чтобы избежать дрейфа времени.

После этого статус задачи устанавливается на «Выполняется» и возвращается мастеру для уведомления о начале выполнения задачи.

Поскольку большинство задач выполняется в операционной системе Linux, требуется обработка клиентов и файлов:

• Обработка арендатора: сначала проверяется, существует ли арендатор. Если нет, то решается, следует ли автоматически создать арендатора на основе конфигурации. Для этого пользователю развертывания необходимо иметь разрешения sudo для переключения на указанного арендатора во время выполнения задачи.
• Конкретные пользователи : для некоторых сценариев не нужно переключать арендаторов, а просто выполнить задачу, используя определенного пользователя. Это также поддерживается системой.

• 
  

После обработки арендатора, воркер создает определенный каталог выполнения. Корневой каталог каталога выполнения настраивается и требует соответствующей авторизации. По умолчанию разрешения каталога установлены на 755.

Во время выполнения задачи могут потребоваться различные файлы ресурсов, например, для извлечения файлов из кластеров AWS S3 или HDFS. Система загружает эти файлы во временный каталог воркера для последующего использования в задаче.

В Apache DolphinScheduler можно заменить переменные параметров. Основные категории включают:

• Встроенные параметры: в первую очередь предполагают замену параметров, связанных со временем и датой.
• Пользовательские параметры: переменные параметров, заданные пользователем в рабочем процессе или задаче, также будут заменены соответствующим образом.

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

Различные типы задач

В Apache DolphinScheduler поддерживаются различные типы задач, каждый из которых применим к различным сценариям и требованиям. Ниже мы представляем несколько основных типов задач и их конкретные компоненты.

Эти компоненты обычно используются для выполнения файлов сценариев, подходящих для различных языков сценариев и протоколов:

• Shell: выполняет сценарии оболочки.
• Python: выполняет скрипты Python.
• SQL: выполняет операторы SQL.
• Хранимая процедура: выполняет хранимые процедуры базы данных.
• HTTP: выполняет HTTP-запросы.

Коммерческая версия (WhaleScheduler) также поддерживает запуск приложений Java путем выполнения пакетов JAR.

Компоненты логической задачи

Эти компоненты используются для реализации логического контроля и управления рабочим процессом:

• Переключатель: Задача условного управления.
• Зависимый: Задача на зависимость.
• Подпроцесс: Подзадача.
• NextLoop (коммерческая версия): задача управления циклом, подходящая для финансовых сценариев.
• Компонент триггера: отслеживает наличие файлов или данных.

Компоненты больших данных

Эти компоненты в основном используются для обработки и анализа больших данных:

• SeaTunnel: соответствует коммерческой версии WhaleTunnel, используемой для интеграции и обработки больших данных.
• AWS EMR: интеграция с Amazon EMR.
• HiveCli: задача командной строки Hive.
• Искра: Задача «Искра».
• Флинк: Задача Флинка.
• DataX: задача синхронизации данных.

Компоненты контейнера

Эти компоненты используются для запуска задач в контейнерной среде:

• K8S: задача Kubernetes.

Компоненты качества данных

Для обеспечения качества данных используется:

• DataQuality: задача проверки качества данных.

Интерактивные компоненты

Эти компоненты используются для взаимодействия со средами науки о данных и машинного обучения:

• Jupyter: задача Jupyter Notebook.
• Zeppelin: Задание «Блокнот Zeppelin».

Компоненты машинного обучения

Эти компоненты используются для управления и выполнения задач машинного обучения:

• Kubeflow: Задача Kubeflow.
• MlFlow: Задача MlFlow.
• Dvc: задача контроля версий данных.

В целом Apache DolphinScheduler поддерживает от трех до четырех десятков компонентов, охватывающих области от выполнения скриптов, обработки больших данных до машинного обучения. Для получения дополнительной информации посетите официальный сайт , чтобы просмотреть подробную документацию.

Тип задачи Абстракция

В Apache DolphinScheduler типы задач абстрагируются в несколько режимов обработки для соответствия различным средам выполнения и потребностям.

Ниже мы подробно рассмотрим процесс абстрагирования и выполнения типов задач.

Worker — это служба JVM, развернутая на сервере. Для некоторых компонентов скрипта (таких как Shell и Python) и локально запущенных задач (таких как Spark Local) они запустят отдельный процесс для выполнения.

На этом этапе работник взаимодействует с этими задачами через идентификатор процесса (PID).

Различные источники данных могут потребовать различных адаптаций. Для задач SQL и хранимых процедур мы абстрагировали обработку для различных источников данных, таких как MySQL, PostgreSQL, AWS Redshift и т. д. Эта абстракция обеспечивает гибкую адаптацию и расширение различных типов баз данных.

Удаленные задачи относятся к задачам, которые выполняются на удаленных кластерах, таких как AWS EMR, кластеры SeaTunnel, кластеры Kubernetes и т. д. Worker не выполняет эти задачи локально; вместо этого он отправляет их на удаленные кластеры и отслеживает их статус и сообщения. Этот режим особенно подходит для облачных сред, где требуется масштабируемость.

Выполнение задачи

Сбор журналов

Различные плагины используют разные режимы обработки, и поэтому сбор журналов различается соответствующим образом:

• Локальные процессы: журналы регистрируются путем мониторинга выходных данных процесса.

• Удаленные задачи: журналы собираются путем периодической проверки статуса задачи и выходных данных из удаленного кластера (например, AWS EMR) и записи их в локальные журналы задач.

  
  

Параметр Замена Переменной

Система сканирует журналы задач, чтобы определить любые переменные параметров, которые необходимо динамически заменить. Например, задача A в DAG может генерировать некоторые выходные параметры, которые необходимо передать в нижестоящую задачу B.

В ходе этого процесса система считывает журналы и заменяет переменные параметров по мере необходимости.

Получение идентификатора задачи

• Локальные процессы: извлекается идентификатор процесса (PID).
• Удаленные задачи: извлекается идентификатор удаленной задачи (например, идентификатор задачи AWS EMR).

Удержание этих идентификаторов задач позволяет выполнять дальнейшие запросы данных и удаленные операции с задачами. Например, когда рабочий процесс останавливается, можно вызвать соответствующий API отмены с использованием идентификатора задачи, чтобы завершить запущенную задачу.

Обработка отказоустойчивости

• Локальные процессы: если узел Worker выйдет из строя, локальный процесс не узнает об этом, и потребуется повторная отправка задачи.
• Удаленные задачи: если задача выполняется на удаленном кластере (например, AWS), статус задачи можно проверить с помощью идентификатора задачи и попытаться перехватить ее. В случае успеха нет необходимости повторно отправлять задачу, что экономит время.

Завершение выполнения задачи

После выполнения задачи требуется выполнить несколько действий по ее завершению:

• Проверка завершения задачи: система проверит, нужно ли отправлять оповещение. Например, для задачи SQL, если результаты запроса вызывают оповещение, система будет взаимодействовать со службой оповещений через RPC для отправки сообщения оповещения.

• Обратная связь по событию: Worker отправит событие завершения задачи (событие завершения) обратно Master. Master обновляет статус задачи в базе данных и переходит к переходу статуса DAG.

• Очистка контекста: Worker удалит контекст задачи, созданный в начале задачи, из памяти. Он также очистит пути к файлам, сгенерированные во время выполнения задачи. Если находится в режиме отладки (режим разработки), эти файлы не будут очищены, что позволит устранять неполадки в невыполненных задачах.

  
  

Посредством этих шагов завершается весь процесс выполнения экземпляра задачи.

Вклад сообщества

Если вас интересует Apache DolphinScheduler и вы хотите внести свой вклад в сообщество разработчиков ПО с открытым исходным кодом, вы можете ознакомиться с нашими правилами внесения вклада.

Сообщество поощряет активные вклады, включая, помимо прочего:

• Сообщение о проблемах, возникших во время использования.
• Подача документации и заявок на внедрение кода.
• Добавление модульных тестов (UT).
• Добавление комментариев к коду.
• Исправление ошибок или добавление новых функций.
• Написание технических статей или участие во встречах.

Руководство для новых участников

Для новых участников вы можете искать проблемы, отмеченные как good first issue в проблемах сообщества GitHub. Эти проблемы, как правило, проще и подходят для пользователей, делающих свой первый вклад.

Подводя итог, мы узнали об общей конструкции Apache DolphinScheduler и подробном процессе выполнения задач Worker.

Надеюсь, этот контент поможет вам лучше понять и использовать Apache DolphinScheduler. Если у вас есть вопросы, не стесняйтесь обращаться ко мне в разделе комментариев.