Dryad. Фреймворк распределенных вычислений

* Статистические данные за 2011 год.

А теперь представьте, что это не Hadoop.

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

1. Dryad. Общие сведения

Dryad – программный фреймворк общего назначения для распределенного исполнения приложений. Dryad является проектом подразделения Microsoft Research. Центральной концепцией проекта Dryad является построение направленного ациклического графа (directed acyclic graph, DAG) для каждой распределенной задачи. Вершины графа представляют собой операции над данными (по сути, программы), а ребра графа — каналы, по которым данные передаются.

Абстракция на основе модели направленного ациклического графа дает возможность эффективно реализовывать планы исполнения большого количества параллельных алгоритмов, итеративных алгоритмов, алгоритмов машинного обучения. Таким образом, единственная (до появления YARN) программная модель map/reduce, реализованная в Hadoop**, по сути является лишь частным случаем модели распределенных вычислений, которую предоставляет Dryad.

Dryad оптимизирован для запуска на среднем или большом вычислительном кластере (от 100 до 10K вычислительных узлов) и нацелен, главным образом, на длительные batch-задания, не требующие частых взаимодействий.

2004… 2008

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

• идея построения такой системы возникла в 2004 году (аналогия с Google MapReduce) у Michael Isard (позже он занял позицию исследователя в Microsoft);
• в 2006 году на одной из «Computer Science»-конференций Dryad презентовал сам Билл Гейтс;
• Dryad адаптирован и используется в Bing, Microsoft AdCenter, Kinect, Windows HPC Server 2008 R2 (последнее, известно достоверно).

2008… 2009

Один из основополагающих документов, описывающих концепции заложенные в Dryad, удостоился награды «Best Paper» на OSDI’08 (USENIX Symposium on Operating Systems Design and Implementation).

В ноябре 2009 года Dryad стал доступен по академической лицензии.

2010… 2013

В 2011 году командой Windows HPC Team анонсирована beta-версия «LINQ to HPC» (Dryad для Windows HPC Server) для соответствующей линейки ОС Windows. В этом же году было объявлено, что LINQ to HPC «не выйдет» из preview-версии (т.е. фактически о прекращении поддержки).

Стоит отметить, что команда Windows HPC Team ничего не заявляла (да и не могла заявить в принципе) о поддержке/развитии Dryad Project. Других заявлений про дальнейшую поддержку Dryad или, напротив, однозначный отказ от поддержки, на протяжении прошлого (2012) и этого года замечено не было.

2. Экосистема Dryad

Проект Dryad состоит из 3-ех ключевых компонентов:

• Dryad — среда исполнения (runtime) распределенных приложений (далее, во избежание неоднозначности, будем называть этот компонент Dryad Runtime);
• DryadLINQ — высокоуровневый язык запросов, базирующейся на программной модели .NET Language Integrated
  Query (LINQ);
• Distributed Storage Catalog(DSC) — распределенная файловая система с конфигурируемой избыточностью.

Ниже подробнее рассмотрим элементы экосистемы Dryad: среду исполнения Dryad runtime и язык запросов DryadLINQ.

3. Dryad Runtime

Dryad runtime представляет собой среду исполнения распределенных приложений, как и Hadoop, беря на себя такие функции как:

• планировка и управление распределенными заданиями;
• управление ресурсами;
• обеспечение отказоустойчивости;
• мониторинг.

Задание в Dryad runtime представляет собой направленный ациклический граф, где вершины представляют собой программы, а ребра графа – каналы данных. Этот логический граф отражается (mapped) исполняемой средой на физические ресурсы, находящиеся в кластере. В общем, случае количество вершин в графе превышает количество физических вычислительных узлов в кластере.

3.1. Каналы данных

Каналы данных, также как и вершины, являются абстракциями и представлены:

• Shared-memory FIFO (intra-machine): наиболее быстрый способ обмена данных между операциями (вершинами графа), но операции должны выполняться в рамках одного вычислительного узла;
• TCP pipes (inter-machine): канал, не требующий доступа к диску (избегаем накладных расходов, связанных с записью/чтением данных с диска); канал возможно использовать только, если операции, между которому данные передаются, доступны в момент времени передачи;
• SMB/NTFS файлы (временные файлы): выходные данные, переданные по каналу, записываются
  на диск, входные данные — считываются; канал по умолчанию.

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

3.2. Модель данных

Модель данных в Dryad представляет собой shared-nothing архитектуру. К достоинствам такой архитектуры традиционно относят масштабируемость, отсутствие необходимости в поддержке отслеживания изменений, реализации сложных транзакций, использовании примитивов синхронизации. Платформа Dryad хорошо подходит для больших статических объемов данных и не подходит для часто изменяемых или потоковых данных.

Dryad рассчитывает получить неизменяемое (immutable) и конечное количество входных данных. Результаты выполнения распределенной программы не будут доступны пока не выполняться все подпрограммы. Логично предположить, что задачи с потоковыми данными являются принципиально неисполнимыми в Dryad.

3.3. Job Manager

Оркестровкой выполнения Dryad-задания занимается менеджер заданий Job Manager (JM). Job Manager содержит специфический для приложения (application-specific) код для создания графа вычисления задач.

Job Manager ответственен за:

1. инициализацию графа вычисления;
2. планирование операций (вершин, в дальнейшем — vertex-операций) так, чтобы физическое аппаратное обеспечение, ассоциированное с этими вершинами, находилось топологически как можно ближе к данным, которые обрабатываются этими вершинами;
3. обеспечение отказоустойчивости;
4. мониторинг выполнения и сбор статистики;
5. динамическую трансформацию графа в соответствие с имеющимися политиками.

Данные приложения пересылаются напрямую от vertex-операции к vertex-операции. Job Manager же ответственен только за инициализацию, планирование и контроль исполнения распределенных заданий. Поэтому JM не является узким местом в исполнении приложения. Кроме того, при пересылке код vertex-операции может быть как переслан от Job Manager, так и с ближайшего вычислительного узла, на котором выполняется аналогичная vertex-операция.

3.4. Name Server. Daemon

За раскрытие информации о доступных вычислительных узлах и их топологическом расположении в кластере отвечает сервер имен Name Server.

На каждом вычислительном узле запущен процесс Daemon, главной целью которого является запуск vertex-операций, присланных Job Manager. Daemon выступает как прокси-объект, поэтому Job Manager имеет возможность узнать статус и этап выполнения vertex-операции, запущенной удаленным Daemon.

Архитектура Dryad и место в этой архитектуре Job Manager, Name Server (NS) и Daemon (PD) приведены ниже.

Источник иллюстрации [3]

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

3.5. Динамическое изменение графа

Job Manager, по аналогии с JobTracker в Hadoop, проводит статическую оптимизацию исполнения. Но в отличие от Hadoop, в Dryad реализована возможность динамической оптимизации исполнения.

Благодаря центральной для Dryad концепции направленного ациклического графа и поддержке механизма обратных вызовов (callback информирует JM об изменении этапа выполнения vertex-операции), граф исполнения способен изменяться в процессе выполнения (runtime). Динамическое изменение позволяет довольно изящно решать следующие задачи:

• деградации пропускной способности сети, когда нескольким узлам нужно передать данные на вход другому узлу (в Hadoop такая деградация наблюдается при пересылке данных между стадиями Combine и Reduce);
  
• простой вычислительных узлов с целью ожидания окончания «медленных» операций (в Hadoop свертка не может начаться пока все map-задания не будут окончены, что приводит к простою map-слотов, и делает лучшее время исполнения программы равным времени исполнения самого «медленного» map-задания).
  

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

• один Daemon может сохранить временный файл с промежуточными результатами в rack’е локальном к vertex-операции, которая эти данные принимает как входные;
• один Daemon может сохранять данные в оперативной памяти узла, если vertex-операция, которая принимает эти данные как входные, запущена на том же вычислительном узле.

3.6. Обеспечение отказоустойчивости

Как уже ранее упоминалось, Daemon представляет собой прокси, поэтому Job Manager имеет возможность узнать статус и этап выполнения vertex-операции, запущенной удаленным Daemon. Если Daemon «упал», то Job Manager об этом узнает:

• по сообщению об ошибке, которые Daemon отправил на JM перед закрытием собственного процесса;
• по просроченному времени heartbeat-сообщения в случае, если при закрытии Daemon никаких диагностических событий на JM не отправил.

После диагностики отказа PD, операция, выполняемая на нем, перевыполняется на другом Daemon.

В документации к Dryad я не нашел информации о том, что произойдет в случаем падения Name Server. Разумно предположить, что при отсутствии heartbeat от NS-процесса Job Manager перезапустит NS на другом узле. На время простоя Name Server часть вычислительной мощности кластера, за раскрытие информации о которой NS отвечает, просто «выпадет».

Также из документации не стало ясно, какие меры предприняты для того, чтобы Job Manager не стал единичной точкой отказа. В случае если каждое распределенное приложение имеет свой собственный Job Manager, то остановка Job Manager не будет приводить к простою целого кластера, как это имеет место у Hadoop при отказе Name Node.

Но наличие отдельных JM-процессов для каждого из распределенных приложений сразу представляет 2 проблемы:

• временные затраты на начальную инициализацию Job Manager;
• проблема разделения ресурсов, Job Manager не может «адекватно» оценить количество свободных ресурсов (CPU, RAM, bandwidth), т.к. не знает сколько еще JM-процессов обслуживает данный физический вычислительный узел.

4. Практика

Dryad доступен бесплатно по академической лицензии [4]. Для запуска фреймворка понадобиться Windows HPC cluster с Microsoft HPC Pack 2008 SP1, 4 Gb RAM, 1Gb Ethernet и 200 Gb свободного места на каждой ноде кластера [1].

Приложения под Dryad можно писать как на С++, так и на C# (разумно предположить, что подойдет любой CLS-совместимый язык).

Инструкция по распределенному выполнению операции для Dryad выглядит следующим образом (InitialReduce, Combine, Initialize, Iterate, Merge – это имена функций, ответственных за соответствующие стадии распределенного выполнения; примеры в листингах считают среднее арифметическое):

///<summary>For iterator-based implementation</summary>  [AssociativeDecomposable("InitialReduce", "Combine")]  public static TOutput H(IEnumerable<TInput> source) { … }

public static double Average(IEnumerable<int> g) {      IntPair final = g.Aggregate(x => PartialSum(x));      if (final.second == 0) return 0.0;      return (double)final.first / (double)final.second;  }   [AssociativeDecomposable("InitialReduce", "Combine")]  public static IntPair PartialSum(IEnumerable<int> g) {      return InitialReduce(g);  }    public static IntPair InitialReduce(IEnumerable<int> g) {  return new IntPair(g.Sum(), g.Count());  }  public static IntPair Combine(IEnumerable<IntPair> g) {      return new IntPair(g.Select(x => x.first).Sum(),          g.Select(x => x.second).Sum());  }

///<summary>For accumulator-based implementation</summary>  [AssociativeDecomposable("Initialize", "Iterate", "Merge")]  public static TOutput H(IEnumerable<TInput> source) { … }

public static double Average(IEnumerable<int> g) {      IntPair final = g.Aggregate(x => PartialSum(x));      if (final.second == 0) return 0.0;      else return (double)final.first / (double)final.second  }   [AssociativeDecomposable("Initialize", "Iterate", "Merge")]  public static IntPair PartialSum(IEnumerable<int> g) {      return new IntPair(g.Sum(), g.Count());  }   public static IntPair Initialize() {      return new IntPair(0, 0);  } public static IntPair Iterate(IntPair x, int r) {      x.first += r;      x.second += 1;      return x;  }  public static IntPair Merge(IntPair x, IntPair o) {      x.first += o.first;      x.second += o.second;      return x;  }

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

5. Ограничения и недостатки

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

Одним из главных ограничений Dryad является сложность адаптации для работы в realtime-режиме и, вероятно, принципиальная невозможность работы с потоковыми данными. Также следует добавить, что фреймворк Dryad покажет хорошую эффективность для batch-заданий, но применение Drayd не будет оправдано для с random-access операций. Дополнительно отмечу потенциальную проблему с единичной точкой отказа [на уровне приложения] при «падении» менеджера заданий Job Manager (возможно такой точки отказа и нет, но из документации не ясно, как эта проблема решается).

Также необходимо понимать, что Dryad только фрейморк распределенных вычислений, поэтому и не стоит ожидать от Dryad:

• поддержки транзакций, потому что это не РСУБД;
• производительности, сопоставимой с вычислениями на GPU, потому что Dryad решает принципиально более широкий класс задач принципиально иными инструментами;
• Open-source лицензии, потому что это Microsoft изначально другой тип продукта (но бесплатный доступ по академической лицензии все же есть);
• быстрой разработки распределенных приложений, работающих в рамках парадигмы map/reduce, потому что Dryad — не Hadoop: модель map/reduce — только частный случай для Dryad, в то время как для Hadoop – это единственно возможная модель выполнения.

Заключение

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

Список источников

[1] The Dryad Project. Microsoft Research.
[2] Y. Yu, P. K. Gunda, M. Isard. Distributed Aggregation for Data-Parallel Computing: Interfaces and Implementations, 2009.
[3] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: Distributed data-parallel programs from sequential building blocks.
In Proceedings of European Conference on Computer Systems (EuroSys), 2007.
[4] Dryad and DryadLINQ Academic Release. Microsoft Research.

* Упорядочены по алфавиту.
** Сравнение в статье происходит исключительно с 1-ой версией платформы Hadoop (т.е. без YARN). Подробное сравнение Dryad с СУБД, GPU-вычислениями и фреймворком Hadoop будет выполнено в заключительной статье цикла про Dryad.