Планировщик заданий DAG 2.0: создание более динамичной и гибкой экосистемы распределенных вычислений

Автор: Чен Инда, старший технический эксперт Alibaba Cloud Intelligence

0. Предисловие

Планировщик заданий и распределенная система выполнения, являющиеся основой для основных возможностей Alibaba в области больших данных, поддерживают большинство требований к вычислениям больших данных платформ больших данных Alibaba Group и Alibaba Cloud. Различные вычислительные механизмы, такие как MaxCompute и PAI, работающие в системе, ежедневно обрабатывают огромные потребности пользователей в вычислениях данных. В экосистеме больших данных Alibaba система Job Scheduler управляет несколькими физическими кластерами внутри и за пределами Alibaba Group с более 100 000 физических серверов и миллионами ядер ЦП и графических процессоров. Распределенная платформа Job Scheduler выполняет более 10 миллионов заданий и обрабатывает эксабайты данных каждый день, что обеспечивает ей лидирующие позиции в отрасли. Одно задание содержит сотни тысяч вычислительных узлов и управляет десятками миллиардов пограничных соединений. Такое количество и масштаб работы подтолкнули распределенную платформу Job Scheduler к постоянному развитию в течение последних 10 лет. С более разнообразными задачами и дальнейшими требованиями к развитию архитектура системы Job Scheduler нуждается в дальнейшем развитии, что является огромной проблемой, но также и прекрасной возможностью для команды Alibaba Job Scheduler. В этой статье описывается, как команда планировщика заданий Alibaba обновила базовую систему планирования и распределенного выполнения за последние два года, чтобы создать DAG 2.0.

1. Фон

1.1 Планировщик заданий DAG или AM

В общем, архитектура распределенной системы, работающей на физических кластерах, может быть разделена на управление ресурсами, распределенное планирование заданий и выполнение нескольких вычислительных узлов, как показано на следующем рисунке. Направленный ациклический граф (DAG) является центральным управляющим узлом каждого распределенного задания, то есть главным приложением (AM). AM часто называют DAG, потому что он координирует выполнение распределенного задания. Работа, выполняемая в современных распределенных системах, может быть описана с помощью планирования DAG и перетасовки данных [1]. По сравнению с традиционной моделью Map-Reduce [2], модель DAG может точно описывать распределенные задания, а также является основой проектирования архитектуры для основных систем больших данных, включая Hadoop 2.0+, Spark, Flink и TensorFlow. Единственная разница между моделью DAG и этими основными системами больших данных заключается в том, раскрывается ли семантика DAG пользователям или разработчикам вычислительных механизмов.

Во всем стеке распределенной системы AM выполняет обязанности в дополнение к выполнению DAG. Как центральный управляющий узел распределенного задания, AM запрашивает вычислительные ресурсы, необходимые для распределенного задания, от нижестоящего Менеджера ресурсов, связывается с вышестоящим вычислительным механизмом и сообщает собранную информацию процессу выполнения DAG. Как единственный компонент, который точно понимает выполнение каждого распределенного задания, AM также точно контролирует и регулирует выполнение DAG. На предыдущем рисунке стека распределенной системы AM является единственным компонентом системы, который должен взаимодействовать почти со всеми распределенными компонентами и играет важную роль во время выполнения задания. В предыдущей системе планировщика заданий AM назывался JobMaster (JM). В этой статье мы называем это DAG или AM.

1.2 Логические и физические графики

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

Чтобы быть конкретным:

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

Есть много эквивалентных методов для преобразования логического графа в физический. Одна из важных обязанностей группы DAG - выбрать правильный метод для преобразования логического графа в физический и гибко настроить граф. Как показано на предыдущем рисунке, отображение логического графа на физический граф выполняется, чтобы ответить на вопросы о физических характеристиках узлов графа и ребер соединения. После ответа на эти вопросы физический граф может быть выполнен в распределенной системе.

1.3. Почему нам нужно было перейти на DAG 2.0?

Как распределенная структура выполнения заданий планировщика заданий, DAG 1.0 был разработан при создании системы Alibaba Cloud Apsara. За последние 10 лет DAG 1.0 поддерживал компании Alibaba, занимающиеся большими данными, и разработал преимущества по сравнению с другими отраслевыми платформами с точки зрения масштабирования системы, надежности и других аспектов. Несмотря на то, что DAG 1.0 непрерывно развивался в течение последних 10 лет, он унаследовал некоторые особенности среды выполнения Map-Reduce и не имеет четкого разделения между логическим и физическим графами. Следовательно, для DAG 1.0 было сложно поддерживать больше динамических функций во время выполнения DAG или в различных режимах вычислений. В настоящее время на платформе MaxCompute используются две полностью разделенные среды распределенного выполнения для автономного режима заданий и режима заданий почти реального времени (Smode) для заданий SQL. В большинстве случаев приоритет отдается производительности или использованию ресурсов. Следовательно, нельзя достичь хорошего баланса между ними.

С обновлениями механизмов MaxCompute и PAI и развитием новых функций возможности распределенных вычислений верхнего уровня постоянно расширяются. Следовательно, нам нужно, чтобы AM был более динамичным и гибким в управлении заданиями, выполнении DAG и других аспектах. Чтобы поддержать развитие вычислительных платформ в течение следующих 10 лет, группа Job Scheduler начала проект DAG 2.0, чтобы полностью заменить компонент JobMaster в DAG 1.0, обновив его код и функции. Это позволило бы ему лучше поддерживать требования к вычислениям верхнего уровня и использовать обновления службы перемешивания и функций FuxiMaster (Resource Manager). Для предоставления корпоративных услуг хорошая среда распределенного выполнения должна обеспечивать поддержку гипермасштабируемых и высокопроизводительных заданий в Alibaba, а также различных требований к масштабированию и режимам вычислений в облаке. Помимо разработки гипермасштабируемых расширенных системных возможностей, нам нужно было упростить использование планировщика заданий в системах больших данных и использовать интеллектуальные и динамические возможности системы для предоставления корпоративных сервисов больших данных, адаптированных к различным масштабам данных и режимам обработки. Это были важные соображения при разработке архитектуры DAG 2.0.

2. Архитектура и общий дизайн DAG 2.0

После исследования компонентов DAG различных распределенных систем в отрасли, включая Spark, Flink, Dryad, Tez и TensorFlow, мы использовали фреймворки Dryad и Tez в качестве ссылок в проекте DAG 2.0. Благодаря основополагающему дизайну, например четкому разделению логических и физических графов, расширяемому управлению конечным автоматом, управлению системой на основе подключаемых модулей и политикам планирования на основе событий, DAG 2.0 может централизованно управлять различными режимами вычислений вычислительной платформы и обеспечивать лучшую динамику. возможности регулировки на разных уровнях во время выполнения задания.

2.1 Динамическое выполнение задания

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

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

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

• Отсутствует статистика фактических входных данных. Даже для структурированных данных, обрабатываемых заданиями SQL, статистика по функциям данных исходных таблиц может быть собрана неправильно. Большинство исходных таблиц не имеют полной статистики данных из-за различных методов хранения данных и отсутствия усовершенствованных статистических методов. Еще сложнее собрать статистику по характеристикам неструктурированных данных, которые необходимо обрабатывать внутри и вне кластера.
• Большое количество черных ящиков пользовательской логики в распределенных заданиях. Универсальная система обработки больших данных должна поддерживать выполнение пользовательской логики. Вычислительные механизмы и распределенные системы не могут понять пользовательскую логику, реализованную с использованием Java или Python, такую ​​как UDF, UDTF, UDJ, Extractor и Outputer, которые обычно используются в заданиях SQL. На протяжении всего рабочего процесса логика пользователя - это черный ящик. Например, более 20% онлайн-заданий SQL на платформе MaxCompute, особенно важных базовых заданий, содержат пользовательский код. Большой объем пользовательского кода в большинстве случаев приводит к тому, что оптимизатор не может предсказать особенности промежуточных данных.
• Неправильные прогнозы оптимизатора обходятся дорого. Когда оптимизатор выбирает план выполнения, он выбирает тот, который оптимизирует производительность, когда данные имеют определенные особенности. Однако, если из-за неверных предположений выбран неправильный план, производительность может ухудшиться или задания могут завершиться сбоем. Обычно слишком много прогнозов, основанных на статической информации, не могут дать идеальных результатов.

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

• Пустая трата физических ресурсов: например, тип ресурса, заранее выбранный вычислительным узлом, может быть неправильным, или может потребоваться большой объем вычислительных ресурсов для обработки недопустимых данных, которые будут отброшены позже.
• Серьезная проблема с длинным хвостом. Например, из-за перекоса или неправильно организованных промежуточных данных вычислительным узлам на этапе может потребоваться обработка очень большого или очень небольшого объема данных.
• Нестабильные задания. Неправильный выбор статического плана оптимизатором может привести к сбою в выполнении плана выполнения.

DAG или AM, как уникальный центральный узел и узел управления расписанием для распределенных заданий, является единственным компонентом распределенной системы, который может собирать и агрегировать связанные данные и динамически корректировать выполнение задания на основе характеристик данных. Этот компонент может настраивать простые графы физического выполнения (например, динамический параллелизм), а также сложный метод перетасовки и метод оркестровки данных. График логического выполнения также может нуждаться в корректировке на основе характеристик данных. Динамическая корректировка логического графика - это новое решение, которое мы изучаем в DAG 2.0.

Четкое физическое и логическое расслоение узлов, ребер и графов, сбор данных на основе событий и управление расписанием, а также реализация функций на основе плагинов позволяют DAG 2.0 собирать данные во время выполнения задания и использовать их для систематического ответа на важные вопросы при преобразовании логической схемы. граф в физический граф. И физические, и логические графики при необходимости могут быть динамическими, а план выполнения может быть соответствующим образом скорректирован. В следующих сценариях реализации мы дополнительно проиллюстрируем, как сильные динамические возможности DAG 2.0 помогают обеспечить более адаптивное и эффективное выполнение распределенных заданий.

2.2 Единая среда выполнения AM или DAG

DAG 2.0 может описывать различные физические характеристики узлов и ребер в абстрактной и иерархической архитектуре узлов, ребер и графов для поддержки различных режимов вычислений. Структуры распределенного выполнения различных механизмов распределенной обработки данных, включая Spark, Flink, Hive, Scope и TensorFlow, являются производными от модели DAG, предложенной Dryad [1]. Мы думаем, что абстрактное и иерархическое описание графов может лучше описать различные модели в системе DAG, такие как автономные, в реальном времени, потоковые и прогрессивные вычисления. Когда DAG 2.0 был впервые реализован, его основной целью было использование одного и того же набора кода и архитектурных систем для унификации нескольких вычислительных режимов, работающих на платформе планировщика заданий, включая автономные задания MaxCompute и задания почти реального времени, задания PAI TensorFlow и другие. задания, не связанные с SQL. В будущем мы планируем шаг за шагом исследовать другие новые режимы вычислений.

2.2.1 Единая структура выполнения для заданий в автономном режиме и почти в реальном времени

Большинство заданий на платформе MaxCompute - это автономные задания SQL (пакетные) и задания почти в реальном времени (смоделированные). По историческим причинам управление ресурсами и выполнение заданий MaxCompute SQL в автономном задании и режимах заданий почти в реальном времени реализованы на основе на двух наборах полностью разделенного кода. Эти два набора кода и функции двух режимов нельзя было повторно использовать. В результате мы не могли найти баланс между использованием ресурсов и производительностью выполнения. В модели DAG 2.0 эти два режима интегрированы и унифицированы, как показано на следующем рисунке.

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

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

Благодаря особенностям автономного режима заданий, таким как приложение ресурсов по запросу и промежуточное хранение данных на диске, интерактивные задания имеют очевидные преимущества в использовании ресурсов, масштабируемости и стабильности. В режиме задания почти в реальном времени резидентные пулы вычислительных ресурсов и групповое планирование (жадное приложение ресурсов) сокращают накладные расходы во время выполнения задания, обеспечивают конвейерную передачу данных и ускоряют выполнение заданий. Однако этот режим не может поддерживать крупномасштабные задания в широком диапазоне из-за особенностей использования ресурсов. DAG 2.0 объединяет эти два режима вычислений в одной архитектуре. Этот метод унифицированного описания позволяет сбалансировать высокий уровень использования ресурсов для автономных заданий и высокую производительность для заданий, выполняемых почти в реальном времени. Когда блок планирования можно свободно настраивать, возможен новый гибридный режим вычислений. Мы называем это режимом исполнения пузырей.

Этот режим пузырьков позволяет пользователям DAG (разработчикам вычислительных механизмов верхнего уровня, таких как оптимизатор MaxCompute) гибко вырезать пузырьковые подграфы в плане выполнения на основе функций плана выполнения и чувствительности пользователей механизма к использованию ресурсов и производительности. Во всплывающей подсказке для повышения производительности используются такие методы, как прямое сетевое соединение и предварительная выборка вычислительного узла. Узлы, которые не используют пузырьковый режим выполнения, по-прежнему работают в традиционном автономном режиме задания. Автономный режим работы и режим работы почти в реальном времени можно описать как два крайних случая пузырькового режима. В новой унифицированной модели вычислительный механизм и среда выполнения могут выбирать различные комбинации использования ресурсов и производительности на основе фактических требований. Типичные сценарии применения включают:

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

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

2.2.2 Поддержка новых режимов вычислений

На базовый дизайн среды выполнения DAG 1.0 глубоко повлиял режим Map-Reduce. Граничное соединение между узлами смешивает различную семантику, включая последовательность планирования, последовательность выполнения и перемешивание данных. В двух узлах, соединенных ребром, нисходящий узел может быть запланирован только после того, как вышестоящий узел запустится, выйдет и сгенерирует данные. Это не относится к некоторым новым режимам вычислений. Например, в режиме вычислений сервера параметров сервер параметров (PS) и работник имеют следующие функции в запущенном процессе:

• PS, как объект, обслуживающий параметры, может работать независимо.
• Рабочий, как потребитель параметров и средство обновления, может работать после запуска PS и должен обмениваться данными с PS во время выполнения процесса.

В этом режиме работы PS и работник имеют отношения зависимости планирования. Однако PS и воркер должны работать одновременно, поэтому нет логики, которая говорит, что воркер может быть запланирован только после выхода PS. Следовательно, в структуре DAG 1.0 PS и worker можно рассматривать только как два независимых этапа для планирования и выполнения. Кроме того, все PS и рабочие могут связываться и координировать друг друга только через прямое соединение между вычислительными узлами и с помощью внешних объектов, таких как ZooKeeper или Apsara Name Service и Distributed Lock Synchronization System. В результате AM или DAG, как центральный узел управления, не вступают в силу, а задания управляются вычислительным механизмом и выполняются посредством координации между вычислительными узлами. Этот децентрализованный метод управления не может обрабатывать сложные сценарии, такие как аварийное переключение.

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

Кроме того, новая модель описания DAG 2.0 допускает более динамическую оптимизацию для заданий TensorFlow или PS на платформе PAI и новые инновационные разработки. В предыдущем динамическом PS DAG вводится управляющий узел. Узел может динамически настраивать приложение ресурсов и параллелизм заданий до и после выполнения рабочей нагрузки PS, чтобы оптимизировать выполнение заданий.

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

3. Интеграция между DAG 2.0 и вычислительными механизмами верхнего уровня.

В качестве основы для распределенной работы вычислительной платформы DAG 2.0 обеспечивает более гибкие и эффективные возможности выполнения для различных вычислительных механизмов верхнего уровня. Эти возможности реализуются на основе подробных вычислительных сценариев. Далее мы будем использовать взаимодействие между DAG 2.0 и различными вычислительными механизмами верхнего уровня, такими как MaxCompute и PAI, чтобы описать, как структура планирования и выполнения DAG 2.0 расширяет возможности вычислений и приложений верхнего уровня.

3.1 Динамическая настройка DAG в рабочем процессе

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

3.1.1 Регулировка динамического параллелизма

Регулировка динамического параллелизма на основе количества промежуточных данных во время выполнения задания является базовой возможностью динамической корректировки DAG. В традиционном статическом MR-задании параллелизм сопоставителя может быть точно определен на основе объема считанных данных. Однако параллелизм редуктора можно только вывести. Как показано на следующем рисунке, когда задание MR объемом 1 ТБ отправляется на обработку, параллелизм редуктора, равный 500, выводится на основе параллелизма сопоставителя, равного 1000. Однако, если сопоставители фильтруют большой объем данных и в конечном итоге генерируют 10 МБ промежуточные данные, 500 параллельных редукторов представляют собой очевидную трату ресурсов. Группа обеспечения доступности баз данных должна динамически регулировать параллелизм редуктора (от 500 до 1) в зависимости от фактического объема данных, созданных сопоставителями.

В реальной реализации выходные данные разделов вышестоящих узлов просто агрегируются в соответствии с коэффициентом настройки параллелизма. Как показано на следующем рисунке, при изменении параллелизма от 10 до 5 может возникнуть ненужный перекос данных.

Регулировка динамического параллелизма DAG 2.0 на основе промежуточных данных всесторонне рассматривает возможность искажения разделов данных и оптимизирует политику динамической корректировки для обеспечения равномерного распределения данных после корректировки. Это эффективно предотвращает перекос данных, вызванный динамическими настройками.

Эта наиболее распространенная настройка параллелизма нисходящих узлов представляет собой интуитивно понятное представление динамических физических графов DAG 2.0. В проекте DAG 2.0 мы также исследуем динамическую настройку параллелизма на основе исходных данных на основе функций обработки данных вычислительного механизма. Например, когда данные двух исходных таблиц объединяются (M1 соединяется с M2 в J) с размером узла, необходимым для представления объема обработанных данных, как показано на следующем рисунке, M1 обрабатывает промежуточную таблицу данных (например, параллелизм M1 является 10), а M2 обрабатывает большую таблицу данных (параллелизм M2 равен 1000.) Наиболее неэффективное планирование основано на параллелизме 10 + 1000. Кроме того, весь вывод M2 должен быть перетасован в J, и, следовательно, параллелизм J также велик (около 1000.)

В этом вычислительном шаблоне M2 нужно только читать и обрабатывать данные, которые могут быть объединены с выходом M1. Это означает, что если выходные данные M1 намного меньше ожидаемых M2 после учета общей стоимости выполнения, мы можем сначала запланировать M1 для вычислений, агрегировать статистику выходных данных M1 в AM или DAG и выбрать только данные, действительные для M2. . Выбор данных, действительных для M2, по сути, представляет собой процесс проталкивания предиката, который может быть совместно определен на основе оптимизатора вычислительной машины и среды выполнения. Это означает, что настройка параллелизма M2 в этом сценарии тесно связана с вычислениями верхнего уровня.

Например, если M2 обрабатывает многораздельную таблицу с 1000 разделами, а последующий участник присоединяется по ключу раздела, теоретически должны быть прочитаны только данные в разделах, которые могут быть объединены с выходными данными M1. Если выходные данные M1 содержат только три ключа разделов исходной таблицы M2, M2 необходимо запланировать только три вычислительных узла для обработки данных из этих трех разделов. В результате параллелизм M2 снижен с 1000 по умолчанию до 3. Это значительно снижает потребление вычислительных ресурсов и ускоряет выполнение задания в несколько раз, сохраняя при этом ту же логическую эквивалентность и правильность вычислений. Оптимизация в этом сценарии включает:

• Параллелизм M2 и объем данных для обработки значительно сокращаются.
• Объем данных M2, которые необходимо перетасовать в J, и вычислительные ресурсы, необходимые для перетасовки, значительно сокращаются.
• Параллелизм J и объем данных для обработки значительно сокращаются.

Как показано на предыдущем рисунке, чтобы гарантировать последовательность планирования от M1 до M2, DAG вводит границу зависимости между M1 и M2. Этот край зависимости представляет только последовательность выполнения и не имеет перетасовки данных. Это отличается от предположения о жестко связанных граничных соединениях и перетасовке данных в традиционной среде исполнения MR или DAG. Это одно из расширений Edge в DAG 2.0.

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

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

SELECT * FROM tpch_lineitem WHERE l_orderkey > 0 LIMIT 5;

В структуре распределенного выполнения план выполнения этой операции разбивает исходную таблицу, планирует необходимое количество преобразователей для чтения всех данных и объединяет выходные данные преобразователя в редуктор для последней операции LIMIT. Если исходная таблица (tpch_lineitem в предыдущем примере) велика и требует 1000 преобразователей для чтения, стоимость планирования для всего процесса распределенного выполнения составляет 1000 преобразователей и один редуктор. В этом процессе есть некоторые аспекты, которые могут быть оптимизированы вычислительной машиной верхнего уровня. Например, каждый сопоставитель может выйти после того, как сгенерирует количество записей, необходимое для LIMIT (LIMIT 5 в предыдущем примере), поэтому ему не нужно обрабатывать все выделенные ему сегменты данных. Однако в структуре статического выполнения 1001 вычислительный узел должен быть запланирован для получения этой простой информации. Это приводит к огромным накладным расходам на выполнение специальных запросов, особенно когда ресурсы кластера ограничены.

Чтобы облегчить такие сценарии LIMIT, DAG 2.0 произвел некоторые оптимизации, основанные на динамических возможностях новой среды выполнения, в том числе:

• Экспоненциальный запуск вышестоящих узлов: если не все вышестоящие узлы сопоставления имеют высокую вероятность работы, пакетная структура DAG планирует узлы сопоставления экспоненциально, то есть 1, 10,… и полностью.
• Раннее планирование нисходящих узлов: записи, созданные вышестоящими узлами, будут переданы в AM как статистические данные в процессе выполнения. Когда AM определяет, что восходящие узлы сгенерировали достаточно записей, он может заранее запланировать нисходящие узлы-редукторы для потребления восходящих данных.
• Раннее завершение работы вышестоящих узлов. Когда нижележащий редуктор определяет, что количество выходных ограничений соответствует требованиям, он немедленно завершает работу. AM может инициировать раннее завершение работы восходящих узлов сопоставителя. В этом сценарии большинство узлов сопоставления не могут быть запланированы. Также всю работу можно выполнить заранее.

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

На следующем рисунке показаны различия до и после оптимизации предыдущего запроса, когда параллелизм сопоставителя равен 4000 в автономном тесте.

После оптимизации время выполнения сокращается более чем в 5 раз, а потребление вычислительных ресурсов сокращается в сотни раз.

Как типичный пример, этот результат автономного теста - что-то вроде идеальной ситуации. Чтобы оценить реальный эффект, мы выбрали онлайн-вакансии с LIMIT, оптимизированные после выпуска DAG 2.0. Статистические результаты через одну неделю:

Оптимизация экономит вычислительные ресурсы, эквивалентные 254,5 ядрам x мин. ЦП + 207,3 ГБ x мин., И сокращает планирование на 4 349 вычислительных узлов для каждого задания.

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

Динамическая настройка параллелизма в различных сценариях и динамическая настройка политики выполнения планирования - это только два примера динамической настройки физических характеристик графа. Регулировка времени выполнения физических функций имеет различные приложения в структуре DAG 2.0, такие как динамическая оркестровка данных и перетасовка для перекоса данных во время выполнения. Далее мы познакомим вас с некоторыми идеями о динамической настройке логических графиков в DAG 2.0.

3.1.2 Настройка динамического логического графика

В распределенных заданиях SQL соединение карты является распространенным методом оптимизации. Если одна из двух таблиц, которые необходимо объединить, небольшая и ее данные могут храниться в памяти одного вычислительного узла, тасование не требуется. Все его данные транслируются на каждый распределенный вычислительный узел, обрабатывающий большую таблицу. Для завершения соединения в памяти устанавливаются хеш-таблицы. Присоединение к карте может значительно уменьшить перетасовку и сортировку больших таблиц и повысить производительность выполнения заданий. Однако, если данные небольшой таблицы не могут быть сохранены в памяти одного вычислительного узла, OOM произойдет, когда хеш-таблицы будут созданы в памяти. В результате все распределенное задание выйдет из строя, и его придется запускать снова. Несмотря на то, что соединение карты может значительно повысить производительность при правильном использовании, в большинстве случаев оптимизатору требуются явные подсказки соединения карты от пользователей для создания плана соединения карты. Кроме того, ни пользователь, ни оптимизатор не могут точно определить ввод таблицы, не являющейся исходной, потому что количество промежуточных данных может быть точно получено только во время выполнения задания.

Соединение карты и соединение сортировки-слияния по умолчанию - это два разных плана выполнения оптимизатора. На уровне DAG они соответствуют двум различным логическим графам. Чтобы поддерживать динамическую оптимизацию на основе функций промежуточных данных во время выполнения задания, инфраструктура DAG должна иметь возможности выполнения динамического логического графа. Это именно то, что предоставляется функцией условного соединения, разработанной в DAG 2.0. Как показано на следующем рисунке, когда алгоритм, используемый для операции соединения, не может быть определен заранее, структура выполнения распределенного планирования позволяет оптимизатору отправлять условный DAG. Эта группа обеспечения доступности баз данных содержит ветви плана выполнения двух методов соединения. Во время выполнения AM динамически выбирает ветвь (план A или план B) для выполнения на основе объема данных восходящего потока. Этот процесс выполнения динамического логического графа гарантирует, что оптимальный план выполнения будет выбран на основе сгенерированных функций промежуточных данных при выполнении задания.

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

3.2 Пузырьковый режим

Пузырьковый режим - это новый режим выполнения заданий, который мы исследовали в архитектуре DAG 2.0. Мы можем настроить размер и расположение пузырьков, чтобы добиться различного баланса между производительностью и использованием ресурсов. Мы дадим несколько интуитивно понятных примеров, которые помогут вам понять, как пузырьковый режим применяется в распределенных заданиях.

Как показано на предыдущем рисунке, задание TPC-H Q21 разделено на три пузырька. Данные могут быть эффективно распределены по конвейеру между узлами, а планирование может быть ускорено с помощью горячих узлов. Потребление ресурсов (ЦП x время) составляет 35% от производительности заданий почти в реальном времени, производительность составляет 96% от производительности заданий почти в реальном времени в режиме единого планирования и примерно на 70% выше, чем у автономных заданий. .

В стандартном тесте TPC-H емкостью 1 ТБ режим пузырьков обеспечивает лучший баланс между производительностью и использованием ресурсов, чем режим автономных заданий и комплексный режим задания почти в реальном времени (групповое планирование). выбранный (размер пузырьков ограничен 500), пузырьковый режим повышает производительность задания на 200%, при этом потребление ресурсов увеличивается всего на 17% по сравнению с автономным пакетным режимом. Пузырьковый режим может достигать 85% производительности задач почти реального времени в режиме «все в одном» с менее чем 40% потребления ресурсов (ЦП x время). Пузырьковый режим обеспечивает хороший баланс между производительностью и использованием ресурсов, обеспечение эффективного использования системных ресурсов. Режим пузыря был применен ко всем онлайн-вакансиям в Alibaba Group. Фактическая производительность онлайн-работы аналогична той, что наблюдалась в тесте TPC-H.

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

4. Динамическая конфигурация и управление ресурсами

В традиционном распределенном задании тип и количество ресурсов (ЦП, ГП и память), требуемых каждым вычислительным узлом, определяются заранее. Однако до запуска распределенного задания сложно выбрать правильный тип и размер вычислительного узла. Даже разработчикам вычислительных машин нужны некоторые сложные правила для оценки правильной конфигурации. В режимах вычислений, в которых конфигурация должна быть отправлена ​​пользователям, пользователям труднее выбрать правильную конфигурацию.

Здесь мы будем использовать задания TensorFlow (TF) на платформе PAI в качестве примеров, чтобы описать, как функция динамической конфигурации ресурсов в DAG 2.0 помогает заданиям TF выбирать подходящие типы графического процессора и улучшать использование графического процессора. По сравнению с процессорами, графические процессоры представляют собой новый тип вычислительных ресурсов с более быстрым обновлением оборудования. Большинство обычных пользователей не знакомы с вычислительными возможностями графических процессоров. В результате пользователи часто выбирают неподходящие типы графических процессоров. Графические процессоры - дефицитный ресурс для онлайн-операций. Количество графических процессоров, поданное для онлайн-тестирования, всегда превышает общее количество графических процессоров в кластере. Поэтому пользователям приходится долго стоять в очереди, чтобы получить ресурсы. Однако фактическая загрузка графического процессора в кластерах в среднем составляет около 20%, что является низким показателем. Разрыв между количеством приложений и фактическим использованием обычно вызван неправильной конфигурацией ресурсов графического процессора, указанной во время настройки задания.

В структуре DAG 2.0 предоставляется дополнительный узел управления вычислениями для ресурсов GPU заданий PAI TF. Для получения дополнительной информации см. Информацию о динамической PS DAG в разделе 2.2.2 «Поддержка новых вычислительных режимов». Узел может запустить алгоритм прогнозирования ресурсов платформы PAI, чтобы определить тип графического процессора, требуемый текущим заданием, а затем отправить динамическое событие в AM, чтобы при необходимости изменить тип графического процессора, запрашиваемый нижестоящим работником. Точное использование ресурсов можно предсказать с помощью пробного прогона на выборочных данных или оптимизации на основе истории (HBO), которая прогнозирует правильное использование ресурсов с учетом характеристик данных, типа алгоритма обучения, а также информации, собранной из исторических заданий.

В этом сценарии управляющий узел и рабочий соединены через параллельный край. Планирование запускается на границе, когда узел управления выбирает тип ресурса и отправляет событие. Эта динамическая конфигурация ресурсов во время выполнения задания улучшает использование графического процессора более чем на 40% кластеров при онлайн-функциональном тестировании.

В качестве важного параметра физических функций настройка динамических функций ресурсов для вычислительных узлов во время выполнения задания широко используется на платформах PAI и MaxCompute. В заданиях MaxCompute SQL можно эффективно спрогнозировать объем ЦП или памяти восходящих и нисходящих узлов на основе функций восходящих данных. Когда происходит OOM, память, запрошенная повторным вычислительным узлом, может быть увеличена для предотвращения сбоев задания. Это новые функции, реализованные в архитектуре DAG 2.0, и мы не будем здесь подробно останавливаться.

5. Разработка и выпуск

Динамические возможности и гибкость групп DAG являются основой распределенной системы. В сочетании с вычислительным механизмом верхнего уровня это может обеспечить более эффективные и точные вычисления верхнего уровня и значительно улучшить производительность и использование ресурсов в определенных сценариях. Это новые функции и режимы вычислений, разработанные в проекте DAG 2.0. Помимо взаимодействия с вычислительным механизмом для более эффективных планов выполнения, гибкое планирование является фундаментальной особенностью выполнения AM или DAG. Политики планирования DAG 2.0 реализованы на основе событийно-ориентированной структуры и гибкой конструкции конечного автомата. На следующем рисунке показаны достижения DAG 2.0 в области базовой инженерной грамотности и производительности.

В качестве примеров используются простейшие задания Map-Reduce (MR). Для этого типа работы мало возможностей для оптимизации выполнения расписания. Выполнение расписания зависит от маневренности системы планирования и всесторонней возможности разблокировки на протяжении всей процедуры обработки. В этом примере также подчеркиваются преимущества производительности планирования DAG 2.0. Преимущества очевидны при большом масштабе работы. В тесте TPCDS, который лучше отражает характеристики производственной нагрузки, даже когда такие функции, как динамический DAG, отключены для обеспечения справедливого сравнения на основе того же плана выполнения и времени выполнения, DAG 2.0 по-прежнему обеспечивает значительное улучшение производительности благодаря эффективному планированию.

Как мы можем использовать фреймворк следующего поколения для беспрепятственного взаимодействия с онлайн-сценариями, чтобы обеспечить крупномасштабное онлайн-приложение? Это важная тема, которая может проиллюстрировать разницу между реальным обновлением производственной системы и небольшой проверкой концепции новой системы. Система Job Scheduler теперь поддерживает десятки миллионов распределенных заданий на платформах обработки больших данных внутри и за пределами Alibaba Group. Чтобы полностью заменить онлайн-распределенную производственную систему, которая поддерживала наши предприятия по работе с большими данными в течение 10 лет, основной компонент распределенного планирования DAG или AM и одновременно предотвращая сбои в существующих сценариях, нам нужно нечто большее, чем просто передовая архитектура и дизайн. Замена двигателя во время работы для обеспечения качественного обновления системы является более сложной задачей, чем проектирование новой системной архитектуры. Для выполнения таких обновлений необходимы стабильная инженерная база и среда тестирования или выпуска. Без такой базы невозможны динамические функции верхнего уровня и новые режимы вычислений.

В настоящее время DAG 2.0 охватывает все автономные задания SQL и задания почти в реальном времени на платформе MaxCompute, а также задания TensorFlow (CPU и GPU) и задания PyTorch на платформе PAI. Он поддерживает десятки миллионов распределенных заданий каждый день и отвечает требованиям Double 11 и Double 12 в 2019 году. Под давлением пиков данных во время Double 11 и Double 12 (более чем на 50% по сравнению с 2018 годом) DAG 2.0 обеспечил на- вывод времени для важных базовых показателей Alibaba на Double 11 и Double 12. Задания большего числа типов, такие как задания межкластерного копирования, переносятся на архитектуру DAG 2.0, а возможности вычислительных заданий были обновлены на основе новой архитектуры. Выпуск платформы DAG 2.0 закладывает прочную основу для разработки новых функций в различных вычислительных сценариях.

6. Перспективы

Основная архитектура Job Scheduler DAG 2.0 направлена ​​на укрепление основы для долгосрочного развития вычислительных платформ Alibaba и поддержку комбинации вычислительных механизмов верхнего уровня с распределенным планированием для реализации различных инноваций и создания новой вычислительной экосистемы. Это обновление архитектуры - важный первый шаг в этом процессе. Для поддержки вычислительных платформ корпоративного уровня и полного спектра различных масштабов и режимов нам необходимо глубоко интегрировать возможности новой архитектуры с вычислительными механизмами верхнего уровня и другими компонентами системы планировщика заданий. В сценариях приложений Alibaba DAG 2.0 сохранил лидирующие позиции по масштабам работы, одновременно реализовав множество инноваций в архитектуре и функциях, в том числе:

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

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

• Обновление архитектуры системы заданий, работающих в режиме, близком к реальному времени. Управление ресурсами не связано с управлением несколькими заданиями, а динамические графики поддерживаются для заданий в режиме, близком к реальному времени.
• Служба ускорения запросов с учетом кеширования для резидентных одноконтейнерных и многослотовых заданий (короткий запрос MaxCompute)
• Управление узлами заданий на основе конечного автомата и интеллектуальный повторный запуск при сбоях
• Метод динамически определяемого перемешивания: используйте рекурсивное перемешивание или другие методы для динамического решения встроенной проблемы крупномасштабных онлайн-заданий.
• Адаптивное и динамическое разделение и агрегирование промежуточных данных для решения различных проблем перекоса данных в распределенных заданиях
• Несколько планов выполнения для заданий TF GPU на платформе PAI
• Прогрессивная и интерактивная динамическая оптимизация через взаимодействие с оптимизатором во время выполнения DAG
• Поддержка императивных языковых функций и взаимодействие с семантикой, такой как IF, ELSE и LOOP, с помощью таких возможностей, как динамический рост DAG

Улучшенные возможности планирования ядра могут предоставить возможности обслуживания корпоративного уровня для различных распределенных вычислительных механизмов верхнего уровня. Преимущества этих улучшенных возможностей планирования вычислений будут переданы предприятиям-клиентам, обслуживаемым через вычислительные механизмы Alibaba Cloud, такие как MaxCompute. За последние 10 лет предприятия Alibaba стремились создать крупнейшую облачную распределенную платформу в отрасли. Чтобы лучше обслуживать Alibaba Group и наших корпоративных пользователей, мы надеемся, что сервисные возможности платформы корпоративного уровня будут постоянно улучшаться с точки зрения производительности, масштабирования и интеллектуальной адаптации. Это упростит использование сервисов распределенных вычислений и обеспечит инклюзивные большие данные.

Для получения дополнительной информации посетите официальный сайт MaxCompute.

Первоисточник:

Получите доступ к экспертному обзору - Подпишитесь на DDI Intel