Аналитические задачи

Для изучения случаев более сложного использования систем обоих типов были разработаны четыре задачи, относящиеся к обработке HTML-документов. Сначала генерировалась коллекция случайных HTML-документов, похожих на те, которые мог бы найти поисковый робот. Каждому узлу назначался набор из 600000 уникальных HTML-документов, каждый со своим уникальным URL. В каждом документе случайным образом с использованием распределения Зипфа генерировались ссылки на другие страницы.

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

CREATE TABLE Documents ( url VARCHAR(100) PRIMARY KEY, contents TEXT );

CREATE TABLE Rankings ( pageURL VARCHAR(100) PRIMARY KEY, pageRank INT, avgDuration INT );

CREATE TABLE UserVisits ( sourceIP VARCHAR(16), destURL VARCHAR(100), visitDate DATE, adRevenue FLOAT, userAgent VARCHAR(64), countryCode VARCHAR(3), languageCode VARCHAR(6), searchWord VARCHAR(32), duration INT );

Генератор файлов создавал уникальные файлы со 155 миллионами записей UserVisits (20 гигабайт на узел) и 18 миллионами записей Rankings (1 гигабайт на узел) в каждом узле. Поля visitDate, adRevenue и sourceIP подбирались равномерным образом из соответствующих диапазонов значений. Все остальные поля подбирались равномерным образом из наборов данных, представляющих собой выборки из реальных данных. Каждый файл данных хранился в каждом узле в виде текстового файла со столбцами, разделяемыми специальными символами.

Аннотация

В настоящее время наблюдается значительный энтузиазм вокруг парадигмы MapReduce (MR) для крупномасштабного анализа данных . Хотя основной поток управления этой инфраструктуры поддерживается в параллельных SQL-ориентированных системах управления базами данных (СУБД) уже более 20 лет, некоторые называют MR кардинально новой вычислительной моделью [, ]. В этой статье описываются и сравниваются обе парадигмы. Кроме того, для обоих видов систем оценивается производительность и сложность разработки. Для этого определяется эталонный тестовый набор, включающий коллекцию задач, которые пропускались на варианте MR с открытыми кодами и на двух параллельных СУБД. Для каждой задачи на кластере из 100 узлов измеряется производительность для разных уровней распараллеливания. Результаты демонстрируют некоторые интересные соотношения. Хотя процесс загрузки данных и настройки выполнения параллельных СУБД длился гораздо дольше, чем для системы MR, наблюдавшаяся производительность этих СУБД была поразительно более высокой. Приводятся соображения о причинах этой значительной разницы в производительности, и рассматриваются реализационные методы, которые следует позаимствовать в будущих системах из обоих видов архитектур.

Архитектурные элементы

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

Аспекты пользовательского уровня

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

Аспекты системного уровня

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

Благодарности

Авторы хотели бы поблагодарить Лакшмиканта Шриниваса (Lakshmikant Shrinivas) на помощь в приведении СУБД-X в работоспособное состояние, а также Криса Олстона (Chris Olston) и рецензентов за их глубокие комментарии и обратную связь. Эта работа частично поддерживалась грантом NSF CluE - 0844013/0844480.

Дополнительные инструментальные средства

Hadoop оснащается рудиментарным Web-интерфейсом, позволяющим пользователям просматривать распределенную файловую систему и отслеживать выполнение заданий. Ко времени проведения тестовых испытаний все дополнительные инструменты, видимо, должны были бы разрабатываться пользователями собственными силами.

С другой стороны, в SQL-ориентированных системах имеется огромное количество инструментальных средств и приложений для построения отчетов и анализа данных. Целые программные индустрии разрабатывают расширения СУБД. Эти инструментальные средства поддерживают (1) визуализацию данных, (2) интеллектуальный анализ данных, (3) репликацию данных и (4) автоматизацию проектирования баз данных. Поскольку технологии MR находятся в стадии зарождения, рынок такого программного обеспечения для MR ограничен; однако по мере роста пользовательской базы многие из существующих SQL-ориентированных средств, вероятно, станут поддерживать системы MR.

Два подхода к крупномасштабному анализу данных

Системы обоих рассматриваемых классов работают на группах компьютеров без общих ресурсов («shared nothing») . Другими словами, система устанавливается на группу независимых машин, каждая из которых располагает локальной дисковой и основной памятью, и все они связаны высокоскоростной локальной сетью. В системах обоих классов параллелизм достигается за счет разбиения любого используемого набора данных на

Функция Map

Для каждой входной пары «ключ/значение» определяется тип записи путем подсчета числа полей, получаемых после расщепления значения по разделителям. Если это запись UserVisits, то к ней применяется фильтр, основанный на предикате вхождения во временной интервал. Эти отобранные записи выводятся с составными ключами вида (destURL, K1), где K1 указывает, что это запись UserVisits. Все записи Rankings выводятся с составными ключами вида (pageURL, K2), где K2 указывает, что это запись Rankings. Выходные записи заново разделяются с использованием поставляемой пользователем функции разделения, которая хэширует только часть URL составного ключа.

Функция Reduce

На входе функция Reduce получает единый отсортированный поток записей в порядке значений URL. Для каждого URL соответствующие ему значения разделяются на два множества на основе компонента-тега составного ключа. Затем функция образует декартово произведение этих двух множеств для вычисления соединения и выводит новые пары «ключ/значение» с ключом sourceIP и значением – кортежем (pageURL, pageRank, adRevenue).

Фаза 2 – На следующей фазе вычисляется суммарное значение adRevenue и среднее значение pageRank на основе значения ключа sourceIP записей, сгенерированных на Фазе 1. На второй фазе функция Reduce используется для того, чтобы собрать в одном узле все записи с одним и тем же значением sourceIP. Для доставки записей прямо в процесс разбиения используется тождественная функция Map из API Hadoop [, ].

Для каждого значения sourceIP эта функция складывает соответствующие значения adRevenue и вычисляет среднее значение pageRank, оставляя запись с максимальным значением суммы adRevenue. Каждый экземпляр Reduce выводит единственную запись с ключом sourceIP и значением – кортежем вида (avgPageRank, totalRevenue).

Фаза 3 – На этой заключительной фазе снова нужно определить только одну функцию Reduce, которая использует выходные данные предыдущей фазы для получения записи с наибольшим значением totalRevenue. Выполняется только один экземпляр этой функции в одном узле – просматриваются все записи, полученные на Фазе 2, и находится целевая запись.

Эта функция обрабатывает все пары «ключ/значение» и отслеживает запись с наибольшим значением поля totalRevenue. Поскольку от API Hadoop совсем не просто узнать общее число записей, которые будут обрабатываться экземпляром Reduce, функция Reduce никак не может узнать, что обрабатывает последнюю запись. Поэтому в своей реализации Reduce авторы переопределили заключительный метод обратного вызова, чтобы MR-программа выводила требуемую запись прямо перед своим завершением.

Рис. 9. Результаты задачи Join

Гибкость

Несмотря на широкое распространение языка SQL, его регулярно ругают за недостаточно удобные выразительные средства. Некоторые люди считают, что в 1970-х сообщество баз данных допустило ошибку, сфокусировавшись на подъязыках данных, которые можно было бы встраивать в любой язык программирования, вместо того чтобы постараться добавить ко всем языкам высокоуровневые средства доступа к данным. К счастью, разработчики новых сред разработки приложений, таких как Ruby on Rails и LINQ , начинают изменять эту ситуацию, используя новые функциональные возможности языков программирования для реализации некоторого паттерна объектно-реляционного отображения. Эти среды программирования позволяют разработчикам извлекать пользу от надежных технологий СУБД, не обременяя себя написанием сложных выражений на SQL.

Сторонники MR утверждают, что SQL не обеспечивает универсальности, свойственной MR. Но почти во всех основных СУБД (коммерческих и категории open-source) теперь обеспечивается поддержка в SQL определяемых пользователями функций, хранимых процедур и определяемых пользователями агрегатов. Все это не обладает универсальностью MR, но способствует повышению уровня гибкости систем баз данных.

Hadoop

Cистема Hadoop является наиболее популярной реализацией с открытыми кодами среды MapReduce, разрабатываемой Yahoo! и Apache Software Foundation . В отличие от реализации Google исходной среды MR, где использовался язык C++, ядро системы Hadoop целиком написано на языке Java. В экспериментах, описываемых в этой статье, использовалась система Hadoop версии 0.19.0, исполняемая в среде Java 1.6.0. Была установлена система с конфигурацией по умолчанию, за исключением следующих изменений, которые было решено внести для улучшения производительности, не отклоняясь от основных принципов ядра MR:

данные хранились в блоках размеров в 256 мегабайт вместо 64 мегабайт, используемых по умолчанию; каждый исполнитель задач JVM запускался с максимальным размером кучи в 512 мегабайт, и DataNode/JobTracker виртуальной машины запускался с максимальным размеров кучи в 1024 мегабайт (при общем объеме основной памяти в 3,5 гигабайт на узел); в Hadoop была разрешена опция «rack awareness» (возможность учета физического размещения узлов при планировании задач) для обеспечения локальности данных в кластере; в Hadoop было разрешено повторно использовать исполнитель задач JVM, а не запускать новый процесс для каждой задачи Map/Reduce.

Кроме того, система была сконфигурирована таким образом, чтобы на каждом узле запускались два экземпляра Map и один экземпляр Reduce.

В среде Hadoop также обеспечивается некоторая реализация распределенной файловой системы Google . При каждом прогоне тестов все входные и выходные файлы сохранялись в распределенной файловой системе Hadoop (Hadoop distributed file system, HDFS). Использовались параметры HDFS по умолчанию с тремя репликами каждого блока и без сжатия. Тестировались и другие конфигурации (без репликации, со сжатием на уровнях блока и записи), но было обнаружено, что в этих условиях тесты выполняются с той же или худшей скоростью (см. п. 5.1.3). После завершения прогона тестов для заданного уровня масштабирования узлов каталоги данных на каждом узле удалялись, и HDFS форматировалась заново, чтобы следующий набор вводных данных реплицировался по уздам равномерно.

Для координации активности в узлах кластера в Hadoop используются центральный трекер заданий и «главный» (master) демон HDFS. Чтобы эти демоны гарантированно не влияли на производительность узлов-обработчиков, оба эти дополнительных компонента среды выполнялись в отдельном узле кластера.

Имеются два способа загрузки данных в распределенную файловую систему Hadoop: (1) использование файловой утилиты с интерфейсом командной строки для выгрузки в HDFS файлов, хранимых в локальной файловой системе, и (2) создание собственной программы загрузки данных, которая записывает данные с использованием внутреннего API ввода-вывода Hadoop. В данном случае не требовалось изменять вводные данные для тестовых MR-программ, и поэтому во всех узлы файлы загружались в HDFS параллельно в виде плоского текста с использованием утилиты командной строки. Хранение данных в такой манере позволяет MR-программам производить доступ к данным с использованием формата данных Hadoop TextInputFormat, в котором в каждом файле ключами являются номера строк, а соответствующие им значения – это содержимое строк. Было установлено, что этот подход приводит к более высокой эффективности как загрузки данных, так и выполнения задач, чем использование сериализованных форматов или средств сжатия данных Hadoop.

В отличие от набора данных задачи Grep, который загружался в HDFS в неизменном виде, наборы данных UserVisits и Rankings требовалось модифицировать, чтобы первый и второй столбцы разделялись символом табуляции, а все остальные поля каждой строки – некоторым уникальным разделителем полей. Поскольку в модели MR нет схем, для обеспечения доступа к различным атрибутам во время выполнения функции Map и Reduce в каждой задаче должны вручную разбивать значение в массив строк, руководствуясь символом-разделителем.

Был написал специальный загрузчик данных, выполняемый параллельно в каждом узле; этот загрузчик считывал строки наборов данных, подготавливал данные, как требовалось, и затем записывал полученный кортеж в плоский текстовый файл в HDFS. Загрузка данных таким способом происходила примерно в три раза медленнее, чем если бы использовалась утилита командной строки, но зато не потребовалось писать для Hadoop специальные обработчики ввода; имеется возможность использовать в MR-программах интерфейс KeyValueTextInputFormat, позволяющий автоматически расщеплять строки текстовых файлов на пары «ключ/значение» по символу табуляции. Было обнаружено, что использование других вариантов форматирования данных, таких как SequenceFileInputFormat или специальные Writable tuples, замедляет и загрузку, и исполнение программы.

Индексация

Во всех современных СУБД для убыстрения доступа к данным используются индексы на основе хэширования или B-деревьев. Если ищется некоторое подмножество записей (например, записи служащих, получающих зарплату больше $100000), то использование подходящего индекса сокращает область поиска. Кроме того, в большинстве систем баз данных для одной таблицы поддерживается несколько индексов. Таким образом, оптимизатор запросов может принять решение о том, какой индекс следует использовать для выполнения данного запроса, или же предпочесть произвести для этого простой последовательный поиск.

По причине предельной простоты модели MR в инфраструктуре MR встроенные индексы не поддерживаются. Программисты должны сами реализовывать все индексы, которые могут понадобиться для ускорения доступа к данным, в своих приложениях. Это непросто, поскольку нужно еще инструментировать механизмы выборки данных инфраструктуры, чтобы в них использовались эти индексы при распространении данных в выполняемые экземпляры Map. И снова, несмотря на то, что каждому MR-программисту приходится заново реализовывать одну и ту же базовую функциональную возможность, эта стратегия приемлема, если не требуется совместное использование индексов несколькими программистами.

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

Инсталляция, конфигурирование и настройка систем

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

Процесс инсталляции СУБД-X был относительно простым. Начальные шаги выполняются через GUI на одном из узлов кластера, а затем через тот же GUI подготавливается некоторый файл, который может передаваться утилите инсталляции, параллельно выполняемой в других узлах для завершения процесса инсталляции. Несмотря на простоту этого процесса, авторы обнаружили, что СУБД-X трудно сконфигурировать, чтобы можно было начать выполнять запросы. Вначале неудачно завершались все операции, кроме самых элементарных. В конце концов, удалось установить, что ядро системы в каждом узле было сконфигурировано с ограничением общего объема выделяемого виртуального адресного пространства. Когда это ограничение превышалось, переставали создаваться новые процессы, и операции в СУБД-X не выполнялись. Авторы решили упомянуть об этой проблеме, хотя она и возникла по их собственной административной ошибке, поскольку их удивило, что средства исчерпывающего зондирования и автоматического конфигурирования СУБД-X не смогли обнаружить наличие этого ограничения. После предыдущего успешного конфигурирования Hadoop это несколько разочаровало автров.

Даже после разрешения этих проблем и наличия работоспособной установки СУБД-X авторам регулярно мешали другие ограничения по памяти. Они пришли к заключению, что значения некоторых параметров, устанавливаемые по умолчанию, являются слишком заниженными для современных систем.
Кроме того, СУБД- X оказалась неэффективной при регулировании распределения памяти при изменении условий. Например, система автоматически расширила буферный пул с 4 мегабайт, принятых по умолчанию, всего лишь до 5 мегабайт (позднее авторы вынудили систему расширить его до 512 мегабайт). Система также выдавала предупреждение о возможной деградации производительности при увеличении размеров динамически распределяемой памяти для сортировки до 128 мегабайт (на самом деле, производительность возросла в 12 раз). Ручное изменение некоторых параметров приводило к автоматическому изменению системой других параметров. Время от времени эта комбинация ручных и автоматических изменений выражалась в такой конфигурации СУБД-X, которая отказывалась загружаться при следующем старте системы. Поскольку для регулирования большинства конфигурационных параметров требовалось наличие работающей СУБД-X, авторы не могли обеспечить для себя устойчивый режим конфигурирования системы, позволяющий восстановить предыдущее состояние.

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

Основным выводом является то, что параллельные СУБД намного труднее инсталлировать и конфигурировать, чем Hadoop. Однако уровни сложности инсталляции и конфигурирования разных параллельных систем баз данных значительно различаются. Одним из небольших преимуществ систем баз данных является то, что требуемая настройка, обычно производится до выполнения запросов, и некоторые настраиваемые параметры (например, размеры буферов сортировки) пригодны для всех задач.В отличие от этого, для Hadoop не только пришлось настраивать систему (например, устанавливать размер блоков), но также и временами настраивать индивидуальную задачу, чтобы она хорошо работала в системе (например, изменять код). Наконец, параллельные СУБД оснащаются инструментами, помогающими в процессе настройки, а в Hadoop настройку приходилось производить методом проб и ошибок. Ясно, что в более зрелую реализацию MR могли бы войти и подобные инструментальные средства настройки.

Исходная MR-задача

Первой тестовой задачей является «задача Grep», взятая из исходной статьи про MapReduce, авторы которой говорят о ней как о «типичном представителе большого подмножества реальных программ, написанных пользователями MapReduce» . Для решения этой задачи каждая система должна просматривать набор данных, состоящий из 100-байтных записей, и производить в них поиск по трехсимвольному шаблону. Каждая запись состоит из уникального ключа, занимающего первые 10 байт, и 90-байтного случайного значения. Искомый шаблон находится по одному разу в последних 90 байтах каждых 10000 записей.

Вводные данные сохраняются в каждом узле в плоских текстовых файлах, по одной записи в каждой строке. Для прогонов тестов на Hadoop эти файлы в неизменном виде загружались прямо в HDFS. Для загрузки данных в Vertica и СУБД-X в каждой из этих систем выполнялись проприетарные команды загрузки, и данные сохранялись с использованием следующей схемы:

CREATE TABLE Data ( key VARCHAR(10) PRIMARY KEY, field VARCHAR(90) );

Задача Grep выполнялась с двумя разными наборами данных. Измерения в исходной статье про MapReduce основывались на обработке 1 Тбайт данных на примерно 1800 узлах, на каждый узел приходилось 5,6 миллионов записей, или примерно 535 Мбайт. В описываемом испытании для каждой системы задача Grep выполнялась на кластерах с 1, 10, 25, 50 и 100 узлами. Общее число записей, обрабатывавшихся на кластере каждого из этих размеров, составляло 5,6 миллионов × число узлов. Данные о производительности каждой из систем не только иллюстрируют то, как они масштабируется при возрастании объема данных, но также позволяют (до некоторой степени) сравнить результаты с исходной системой MR.

В то время как в первом наборе данных размер данных в расчете на один узел поддерживается таким же, как в исходном тесте MR, и изменяется только число узлов, во втором наборе данных общий его размер устанавливается таким же, как в исходном тесте MR (1 Тбайт), и эти данные поровну разделяются между меняющимся числом узлов. Эта задача позволяет сравнить, насколько хорошо масштабируется каждая система при возрастании числа доступных узлов.

Поскольку для Hadoop требуется в целом 3 Тбайт дисковой памяти, чтобы сохранять в HDFS три реплики каждого блока, пришлось запускать этот тест только на кластере с 25, 50 и 100 узлами (при наличии менее чем 25 узлов для хранения 3 Тбайт данных не хватает дисковой памяти).

Команды SQL

Поиск по шаблону для заданного поля выражается в SQL виде следующего запроса:

SELECT * FROM Data WHERE field LIKE ‘%XYZ%’;

В обеих SQL-ориентированных системах отсутствовал индекс на атрибуте field, и поэтому для выполнения запроса требовался полный просмотр таблицы.

В обеих СУБД задача выборки исполнялась с использованием следующего простого оператора SQL:

SELECT pageURL, pageRank FROM Rankings WHERE pageRank > X;

В отличие от сложной MR-программы, описываемой ниже, для выполнения задачи на СУБД требуются только два довольно простых запроса. Первый оператор создает временную таблицу и использует ее для сохранения результатов оператора SELECT, который выполняет соединение таблиц UserVisits и Ranking и вычисляет агрегаты. После заполнения этой таблицы тривиальным образом используется второй запрос, выводящий запись с наибольшим значением поля totalRevenue.

SELECT INTO Temp sourceIP, AVG(pageRank) as avgPageRank, SUM(adRevenue) as totalRevenue FROM Rankings AS R, UserVisits AS UV WHERE R.pageURL = UV.destURL AND UV.visitDate BETWEEN Date(‘2000-01-15’) AND Date(‘2000-01-22’) GROUP BY UV.sourceIP;

SELECT sourceIP, totalRevenue, avgPageRank FROM Temp ORDER BY totalRevenue DESC LIMIT 1;

Для выполнения этой задачи в параллельной СУБД требуется определяемая пользователем функция F, которая разбирает содержимое каждой записи таблицы Documents и записывает в базу данных найденные URL. Эту функцию можно написать на языке общего назначения, и она, по существу, идентична программе Map, обсуждаемой ниже. С использованием этой функции F во временную таблицу записывается список URL, а затем выполняется простой запрос, вычисляющий число входящих ссылок:

SELECT INTO Temp F(contents) FROM Documents;

SELECT url, SUM(value) FROM Temp GROUP BY url;

Несмотря на простоту предложенной UDF, авторы обнаружили, что на практике ее затруднительно реализовать в СУБД. Для СУБД-X MR-программа, использовавшаяся в Hadoop, транслировалась в эквивалентную C-программу, в которой для поиска ссылок в документе использовалась библиотека регулярных выражений POSIX. Для каждого URL, обнаруживаемого в документе, эта UDF возвращает серверу баз данных новый кортеж (URL, 1). Изначально авторы намеревались хранить каждый документ в СУБД-X в виде символьного BLOB, а затем выполнять UDF над каждым документом полностью внутри базы данных, но так сделать не удалось из-за известной ошибки в их версии системы. Взамен этого UDF была модифицирована для открытия каждого HTML-документа на локальном диске и обработки его содержимого таким образом, как если бы он хранился в базе данных. Хотя это похоже на подход, который пришлось применять для Vertica (см. ниже), UDF в СУБД-X выполняется не как внешний процесс по отношению к базе данных, и для ее выполнения не требуются какие-либо средства массовой загрузки для импорта извлекаемых URL.

В Vertica в настоящее время не поддерживаются UDF, и поэтому авторы были вынуждены реализовать данную тестовую задачу в две фазы. На первой фазе использовалась модифицированная версия UDF для СУБД-X для извлечения URL из файлов, но выходные данные писались в файлы локальной файловой системы каждого узла. В отличие от СУБД-X, эта программа выполнялась в отдельном процессе вне системы баз данных. Затем на каждом узле содержимое этих файлов загружалось в некоторую таблицу с использованием инструментов массовой загрузки Vertica. После завершения этой работы выполнялся описанный выше запрос для вычисления счетчика ссылок для каждого URL.

Конфигурация узлов

Все три системы развертывались на кластере со 100 узлами. В каждом узле имелся один процессор Intel Core 2 Duo, работавший на частоте 2,40 Ггц, с 4 гигабайтами основной памяти и двумя 250-гигабайтными дисками SATA-I. Все узлы работали под управлением ОС Red Hat Enterprise Linux 5 (версия ядра 2.6.18). По данным hdparm дисковая подсистема обеспечивала пропускную способность в 7 Гбайт/сек для кэшированного чтения (cached read) и около 74 Мбайт/сек для буферизованного чтения (buffered read). Для соединения узлов использовались коммутаторы Cisco Catalyst 3750E-48TD. В таком коммутаторе имелись порты гигабайтного Ethernet для каждого узла и внешняя коммутирующая матрица (switching fabric) с пропускной способностью в 128 Гбайт/сек . На каждый коммутатор приходилось 50 узлов. Коммутаторы связывались с использованием технологии Cisco StackWise Plus, что создавало между коммутаторами кольцо с пропускной способностью в 64 Гбайт/сек. Трафик между узлами, подключенными к одному и тому же коммутатору, был полностью локальным для этого коммутатора и не влиял на трафик в кольце.

Литература

Hadoop. http://hadoop.apache.org/. Hive. http://hadoop.apache.org/hive/. Vertica. http://www.vertica.com/. Y. Amir and J. Stanton. The Spread Wide Area Group Communication System. Technical report, 1998. R. Chaiken, B. Jenkins, P.-A. Larson, B. Ramsey, D. Shakib, S. Weaver, and J. Zhou. Scope: easy and efficient parallel processing of massive data sets. Proc. VLDB Endow., 1(2):1265–1276, 2008. Cisco Systems. Cisco Catalyst 3750-E Series Switches Data Sheet, June 2008. J. Cohen, B. Dolan, M. Dunlap, J. M. Hellerstein, and C. Welton. MAD Skills: New Analysis Practices for Big Data. Under Submission, March 2009. J. Dean and S. Ghemawat. MapReduce: Simplified Data Processing on Large Clusters. In OSDI ’04, pages 10–10, 2004. D. J. DeWitt and R. H. Gerber. Multiprocessor Hash-based Join Algorithms. In VLDB ’85, pages 151–164, 1985. D. J. DeWitt, R. H. Gerber, G. Graefe, M. L. Heytens, K. B. Kumar, and M. Muralikrishna. GAMMA - A High Performance Dataflow Database Machine. In VLDB ’86, pages 228–237, 1986. S. Fushimi, M. Kitsuregawa, and H. Tanaka. An Overview of The System Software of A Parallel Relational Database Machine. In VLDB ’86, pages 209–219, 1986. S. Ghemawat, H. Gobioff, and S.-T. Leung. The Google File System. SIGOPS Oper. Syst. Rev., 37(5):29–43, 2003. M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly. Dryad: Distributed Data-parallel Programs from Sequential Building Blocks. In EuroSys ’07, pages 59–72, 2007. E. Meijer, B. Beckman, and G. Bierman. LINQ: reconciling object, relations and XML in the .NET framework. In SIGMOD ’06, pages 706–706, 2006. C. Olston, B. Reed, U. Srivastava, R. Kumar, and A. Tomkins. Pig Latin: A Not-So-Foreign Language for Data Processing. In SIGMOD ’08, pages 1099–1110, 2008. J. Ong, D. Fogg, and M. Stonebraker. Implementation of data abstraction in the relational database system ingres. SIGMOD Rec., 14(1):1–14, 1983. D. A. Patterson. Technical Perspective: The Data Center is the Computer. Commun. ACM, 51(1):105–105, 2008. R. Rustin, editor. ACM-SIGMOD Workshop on Data Description, Access and Control, May 1974. M. Stonebraker. The Case for Shared Nothing. Database Engineering, 9:4–9, 1986. M. Stonebraker and J. Hellerstein. What Goes Around Comes Around. In Readings in Database Systems, pages 2–41. The MIT Press, 4th edition, 2005. D. Thomas, D. Hansson, L. Breedt, M. Clark, J. D. Davidson, J. Gehtland, and A. Schwarz. Agile Web Development with Rails. Pragmatic Bookshelf, 2006.

MapReduce

Одним из привлекательных качеств модели программирования MapReduce является ее простота: MR-программа состоит всего из двух функций, называемых

Модель отказов

Как обсуждалось ранее, при отсутствии поддержки транзакций MR обладает возможностью восстанавливаться после сбоев в середине выполнения запроса с применением метода, не свойственного параллельным системам баз данных. Поскольку параллельные СУБД со временем будут устанавливаться на кластерах более крупного размера, вероятность аппаратного сбоя в середине обработки запроса будет возрастать. Поэтому для долговременно обрабатываемых запросов может оказаться важно реализовать такую модель устойчивости к сбоям. Хотя повышение уровня отказоустойчивости СУБД, очевидно, является правильной идеей, авторы сомневаются в целесообразности выделения для вычислений огромных вычислительных кластеров и применения подходов «грубой силы». Более сложное программное обеспечение могло бы поддерживать ту же самую обработку с применением гораздо меньшего объема аппаратуры, потреблением гораздо меньшей энергии и меньшего времени, позволяя обойтись без сложной модели отказоустойчивости. Кластеры с многотысячными узлами от Google, Microsoft и Yahoo! потребляют громадную энергию, и как показывают результаты авторов, для многих задач обработки данных параллельные СУБД часто могут обеспечить такую же производительность при использовании меньшего числа узлов. По существу, желательный подход состоит в использовании высокопроизводительных алгоритмов с применением умеренного параллелизма, а не подходов грубой силы на гораздо более крупных кластерах.

Модель программирования

В течение 1970-х в исследовательском сообществе баз данных велись бурные дебаты между сторонниками реляционного подхода и приверженцами Codasyl . Основным предметом спора было то, как следует писать программу для доступа к данным в СУБД:

формулируя свою потребность, а не представляя алгоритм ее удовлетворения (реляционный подход), или

представляя алгоритм доступа к данным.

В конце концов, победила первая точка зрения, и прошедшие 30 лет доказали значимость реляционных систем баз данных. Программы на языках высокого уровня, таких как SQL, проще писать, проще изменять и проще понимать новому человеку. Codasyl критиковался за то, что в этом подходе предлагался «язык ассемблера для доступа к базам данных». MR-программирование в чем-то аналогично Codasyl-программированию: человека вынуждают писать алгоритмы на языке низкого уровня для выполнения манипуляций над записями. С другой стороны, для многих людей, привыкших к программированию на процедурных языках, таких как C/C++ или Java, описание задач на декларативном языке, подобном SQL, может быть затруднительно.

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

Обсуждение

Обсудим теперь более общие вопросы, касающиеся результатов тестовых испытаний, и прокомментируем отдельные аспекты каждой системы, которые не выражаются «голыми» цифрами. В описанном выше тестовом наборе СУБД-X и Vertica выполняют большинство задач на всех уровнях масштабирования намного быстрее, чем Hadoop. В следующих подразделах причины этого значительного различия в производительности обсуждаются гораздо подробнее, чем в предыдущем разделе.

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

В средах MR поддерживается более сложная модель обработки сбойных ситуаций, чем в параллельных СУБД. Хотя в обоих классах систем используется некоторая форма репликации для обработки отказов дисков, в подходе MR используются гораздо более искушенные методы обработки отказов узлов при выполнении MR-вычислений. Если в системе MR из-за отказа узла не удается выполнить некоторую единицу работы (т.е. обработку блока данных), то планировщик MR может автоматически перезапустить эту задачу в резервном узле. Эта гибкость частично следует из того, что выводные файлы фазы Map локально материализуются, а не передаются в узлы, выполняющие задачи Reduce, в потоковом режиме. Аналогично, в конвейерах заданий MR, один из которых описывается в п. 4.3.4, промежуточные результаты на каждом шаге материализуются в файлы. Это отличается от подхода параллельных СУБД, в которых в сбойных ситуациях перезапускаются более крупные единицы работы (т.е. транзакции). Этот подход частично обосновывается тем, что СУБД по мере возможности избегают сохранения на диске промежуточных результатов. Поэтому, если во время выполнения какого-либо сложного запроса происходит отказ какого-либо одного узла, то необходимо повторить выполнение всего запроса целиком.

Параллельные СУБД

Системы баз данных, способные функционировать в кластерах узлов без общих ресурсов, существуют с конца 1980-х. Все эти системы поддерживают стандартные реляционные таблицы и SQL, и, таким образом, тот факт, что данные хранятся в нескольких машинах, является прозрачным для конечного пользователя. Многие из этих систем основывались на пионерских исследовательских результатах, полученных при выполнении проектов параллельных СУБД Gamma и Grace . Возможность параллельного исполнения обеспечивается двумя ключевыми аспектами: (1) почти все (или даже все) таблицы разделяются по узлам кластера, и (2) в системе используется оптимизатор, транслирующий команды SQL в план запроса, выполнение которого распределяется по нескольким узлам. Поскольку от программистов требуется только указание своей цели на высокоуровневом языке, они не обременяются деталями уровня хранения данных, такими как варианты индексации или стратегии выполнения соединений.

Рассмотрим SQL-команду для фильтрации записей таблицы T1 по некоторому предикату, ее соединения со второй таблицей T2 и вычисления агрегата на результате соединения. Основная схема обработки этой команды на параллельной СУБД включает три фазы. Поскольку таблица T1 хранится в базе данных, уже разделенная по некоторому атрибуту в некотором наборе узлов, сначала в этих узлах параллельно выполняется подзапрос фильтрации аналогично тому, как выполняется фильтрация в функции Map. Далее, в зависимости от размера таблиц применяется один из двух распространенных параллельных алгоритмов соединения. Например, если в таблице T2 содержится небольшое число записей, СУБД могла бы реплицировать ее по всем узлам при начальной загрузке данных. Это позволяет параллельно выполнять соединение во всех узлах. После этого в каждом узле вычисляется агрегат с использованием своей части результата соединения. И, наконец, для вычисления окончательного результата по этим частичным агрегатам требуется завершающий шаг «свертки» (roll-up) .

Если таблица T2 имеет большой размер, то ее содержимое будет распределено между несколькими узлами. Если эти таблицы разделены не по тем атрибутам, которые используются в соединении, система будет должна выполнить хэширование как T2, так и отфильтрованного варианта T1 с использованием некоторой общей хэш-функции. Перераспределение по узлам T2 и отфильтрованного варианта T1 аналогично обработке, которая происходит после завершения функции Map и до начала выполнения Reduce. Как только в каждом узле будут иметься необходимые данные, в них будут выполнены соединение с хэшированием и предварительное вычисление агрегатной функции. На последнем шаге опять потребуется произвести вычисление свертки для получения окончательного результата.

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

Поддержка схемы

Для параллельных СУБД требуется, чтобы данные подстраивались под реляционную парадигму строк и столбцов. В отличие от этого, в модели MR не требуется, чтобы файлы данных соответствовали какой-либо схеме, определенной с использованием реляционной модели данных. Другими словами, MR-программист свободен структурировать свои данные любым образом или даже оставить их вообще не структурированными.

Можно подумать, что отсутствие жесткой схемы автоматически делает MR предпочтительным вариантом. Например, SQL часто критикуют за то, что от программиста требуется спецификация «формы» данных на языке определения данных. С другой стороны, MR-программистам часто приходится писать свои собственные парсеры, чтобы извлечь соответствующую семантику из своих входных записей, а для этого приходится выполнить, по крайней мере, не меньший объем работы. Но при отказе от использования схемы для крупных наборов данных имеются и другие потенциальные проблемы.

Какая бы структура не существовала во вводных файлах MR, она должна быть встроена в программы Map и Reduce. В существующих реализациях MR поддерживается встроенная функциональная возможность манипулирования простыми форматами «ключ/значение», но программисты вынуждены писать явный код для поддержки более сложных структур данных, например, составных ключей. Возможно, этот подход является приемлемым, если набор данных MR не используется в нескольких приложениях. Однако если такое совместное использование данных имеет место, второму программисту придется разбираться в коде, написанном первым программистом, чтобы понять, как следует обрабатывать вводной файл. Во всех SQL-ориентированных СУБД используется более правильный подход, при котором схема отделяется от приложения и хранится в системных каталогах, к которым можно адресовать запросы.

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

Как только программисты договорятся о структуре данных, что-то или кто- то должен гарантировать, что при любых добавлениях или обновлениях данных не нарушается целостность или другие высокоуровневые ограничения (например, зарплата служащих должна быть неотрицательной). Такие условия должны быть известны всем программистам, модифицирующим набор данных, и должны явно ими соблюдаться. В инфраструктуре MR и распределенной системе хранения данных, на которых MR основывается, отсутствует знание этих правил, и это позволяет легко повредить вводные данные. Но опять же, если отделить такие ограничения от приложения и возложить их поддержку на управляющую систему, как это делается во всех SQL-ориентированных СУБД, то целостность данных будет обеспечиваться без дополнительной работы программистов.

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

Молодежи свойственно увлекаться новыми идеями.

Молодежи свойственно увлекаться новыми идеями. Идея MapReduce, выдвинутая и реализованная сначала Google, а потом и сообществом open source в проекте Hadoop почти мгновенно овладела молодыми массами. Причем даже теми представителями компьютерной молодежи, которые получили хорошее образование и последующий практический опыт в области систем управления базами данных. Мне неоднократно приходилось слышать от молодых коллег, что они считают достоинствами MapReduce отсутствие схемы данных (в том числе, и отсутствие поддержки типов данных) и даже потребность в явном программировании конструкций, которые испокон веков поддерживались в СУБД на уровне высокоуровневых языковых конструкций языка SQL. Понятно, что дополнительным стимулом к применению MapReduce была привязка этой технологии к «облачным» вычислениям, возможность практически бесплатно арендовать виртуальный кластер с большим числом узлов и развернуть на нем свою MapReduce программу, почти автоматически достигнув громадной производительности своего приложения.
До поры до времени представители старшего и среднего поколений сообщества баз данных ограничивались ворчанием в адрес MapReduce, что, в свою очередь, еще больше привлекало молодежь к использованию соответствующих средств. Действительно, раз «старики» ворчат, значит, они просто не понимают, что средства управления данными их поколений просто устарели, и нужно переходить к использованию новых, прогрессивных технологий.
И вот, наконец, ворчание стариков (а больше других ворчали Майкл Стоунбрейкер и Дэвид Девитт) выразилось в инициировании ими чрезвычайно интересного проекта по практическому сравнению технологии MapReduce с технологиями параллельных СУБД категории sharing nothing. Результатам этого проекта и посвящается статья, пересказ которой предлагается вашему вниманию.
Как мне кажется, статья написана предельно объективно. В ней подчеркивается ряд достоинств MapReduce. Некоторые из них кажутся мне сомнительными (например, то, что написание явного кода приложений оказывается проще использования функционально эквивалентных конструкций SQL), но это уже вопросы вкуса. Но основной итог статьи состоит в том, что на простых аналитических задачах параллельные СУБД просто кладут на лопатки Hadoop. И авторы показывают, что здесь дело совсем не в убогости этой реализации (хотя и отмечаются пути ее совершенствования), а в архитектурных недостатках MapReduce.
Финал статьи написан очень мирно, типа «ребята, давайте жить дружно». Другими словами, не отрицайте достижений технологии баз данных, а старайтесь использовать эти достижения в новых технологиях. А сообщество баз данных постарается, в свою очередь, перенять те аспекты технологии MapReduce, которых не достает в современных СУБД.
Статья информативна и увлекательна. Желаю вам приятного чтения.
Сергей Кузнецов

Программа MapReduce

MR-программа состоит из одной функции Map, которая получает одиночную запись, уже расщепленную в соответствующую пару «ключ/значение», и выполняет сопоставление значения с подстрокой. Если поиск подстроки успешно завершается, то функция Map просто выводит полученную пару «ключ/значение» в HDFS. Поскольку нет никакой функции Reduce, выходные данные каждого экземпляра функции Map образуют окончательный результат программы.

Рис. 4. Результаты задачи Grep – набор данных с 535 мегабайтами на узел

Рис. 5. Результаты задачи Grep – набор данных с 1 терабайтом на кластер

В MR-программе использовалась одна функция Map, которая расщепляла входное значение на основе поля-разделителя и выводила значения pageURL и pageRank в качестве новой пары «ключ/значение», если значение pageRank превышало заданное пороговое значение. Для выполнения этой задачи не требуется функция Reduce, поскольку все значения pageURL в наборе данных Rankings уникальны во всех узлах.

Рис. 6. Результаты задачи Selection

Поскольку в модели MR отсутствует внутренняя возможность соединять два или несколько разных наборов данных, MR-программу, реализующую задачу соединения, приходится разбивать на три разные фазы. Все эти фазы реализуются вместе, как одна MR-программа в Hadoop, но следующая фаза не начинает выполняться, пока не завершится предыдущая.

Фаза 1 – На первой фазе отсеиваются записи UserVisits, которые выходят за пределы требуемого временного интервала, и оставшиеся записи соединяются с записями из файла Rankings. Вначале MR-программе в качестве входных данных даются все файлы данных UserVisits и Rankings.

Для обеспечения соответствия с моделью MR, в которой все данные должны определяться в терминах пар «ключ/значение», каждый HTML-документ разбивается на строки и передается функции Map в виде последовательности пар, в которых содержимое строки является значением, а номер строки – ключом. Затем функция Map использует некоторое регулярное выражение для нахождения всех URL в каждой строке. Для каждого находимого URL функция выводит этот URL и целое значение 1 в качестве новой пары «ключ/значение». При наличии этих записей функция Reduce затем просто подсчитывает число значений с данным ключом и выводит URL и вычисленный счетчик входящих ссылок как окончательный результат программы.

Рис. 10. Результаты задачи UDF Aggregation

Простота использования

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

Также очень важно проанализировать способ написания программистом запросов. MR-программы в Hadoop в основном пишутся на Java (хотя существуют и другие возможности). Большая часть программистов в большей степени знакома с объектно-ориентированным, императивным программированием, чем с другими языковыми технологиями, такими как SQL. Вместе с тем, SQL изучается во многих программах для студентов младших курсов, и этот язык является довольно системно-независимым – авторы смогли использовать с небольшими изменениями одни и те же SQL-команды для СУБД-X и Vertica.

Авторы установили, что, вообще говоря, для подготовки MR-программы и ее запуска в среде Hadoop требуется меньше усилий, чем при использовании других систем. Не требуется конструировать схему или регистрировать определяемые пользователями функции, чтобы можно было начать обрабатывать данные. Однако после получения начальных результатов авторы расширяли набор тестовых задач, что потребовало добавления новых столбцов в наборах данных. Для обработки этих новых данных пришлось модифицировать существующий MR-код и заново тестировать каждую MR-программу, чтобы убедиться, что все они работают при новых предположениях о схеме данных. Кроме того, некоторые методы API Hadoop были исключены в новых версиях системы, на которые пришлось перейти авторам, что опять же потребовало переписи частей программ. В отличие от этого, после построения исходных приложений на основе SQL, авторам вообще не пришлось модифицировать их код, не считая нескольких изменений в схеме тестовых данных.

Авторы полагают, что, хотя разработчикам может оказаться проще начать работать с MR, поддержка MR-программ со временем приносит разработчикам приложений существенное беспокойство. Как уже говорилось в подразделе 3.1, трудно использовать один и тот же MR-код в разных условиях или с разными наборами данных, поскольку для данных, используемых в модели MR, отсутствует явное представление схемы.

Распределение данных

Общепринятая точка зрения относительно крупномасштабных баз данных состоит в том, что вычисления следует выполнять рядом с данными, и что не следует пересылать объемные данные к месту их обработки. Другими словами, лучше послать по сети небольшую программу на узел с данными, чем импортировать с него большой объем данных. В параллельных СУБД с пользой применяется знание о распределении данных и их местоположении: параллельный оптимизатор запросов старается балансировать вычислительную нагрузку, минимизируя при этом объем данных, передаваемых по сети, которая соединяет узлы кластера.

За исключением начального планирования расположения экземпляров Map, MR-программист должен выполнять эти задачи вручную. Например, предположим, что пользователь пишет MR-программу из двух частей для обработки набора документов. Сначала функция Map сканирует документы и создает гистограмму часто встречающихся слов. Затем эти документы передаются функции Reduce, которая группирует файлы по именам сайтов, откуда они происходят. Теперь этот или другой пользователь хочет на основе этой работы найти сайты, содержащие документы с более чем пятью вхождениями слов «Google» или «IBM». При наивной реализации этого запроса, в которой Map выполняется над собранной статистикой, фильтрация выполняется после того, как статистика подсчитана для всех документов и передана обработчикам Reduce, даже если заданному условию удовлетворяет только небольшая часть документов.

Аналогичное вычисление производят следующие представление и оператор выборки на языке SQL:

CREATE VIEW Keywords AS SELECT siteid, docid, word, COUNT(*) AS wordcount FROM Documents GROUP BY siteid, docid, word;

SELECT DISTINCT siteid FROM Keywords WHERE (word = ‘IBM’ OR word = ‘Google’) AND wordcount > 5;

В современных СУБД второй запрос был бы переписан таким образом, чтобы в разделе FROM ссылка на таблицу Keywords была заменена определением представления. После этого оптимизатор может выбрать план выполнения запроса, в котором раздел WHERE запроса будет применяться к таблице Documents до вычисления COUNT, что позволит существенно сократить объем вычислений. Если документы распределены по нескольким узлам, то этот фильтр можно применить на каждом узле до группирования документов по сайтам, к которым они относятся, что существенно уменьшит объем данных, передаваемых по сети.

Разделы

, которые для обеспечения параллельной обработки размещаются в разных узлах. В этом разделе приводится обзор того, как в этой среде функционируют MR и традиционные параллельные СУБД.

Reduce

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

Функция Map читает некоторый набор «записей» из входного файла, производит любые требуемые фильтрации и/или трансформации и выводит некоторый набор промежуточных записей в форме новых пар «ключ/значение». По мере того как функция Map производит эти выходные записи, функция «расщепления» (split) разделяет их на R непересекающихся бакетов, применяя некоторую функцию к значению ключа каждой записи. Эта функция расщепления обычно является хэш-функцией, хотя можно использовать любую детерминированную функцию. Каждый сформированный бакет записывается на локальный диск обрабатывающего узла. Функция Map завершается, произведя R выходных файлов, по одному на каждый бакет.

В общем случае имеется несколько экземпляров функции Map, выполняющихся в разных узлах вычислительного кластера. Термин экземпляр (instance) здесь означает индивидуальный выполняемый вызов либо функции Map, либо функции Reduce. Каждому экземпляру Map планировщиком MR назначается для обработки отдельная часть входного файла. Если имеется M таких отдельных частей, то каждая из M задач Map образует R файлов в дисковой памяти, т.е. всего образуется M × R файлов Fij, 1 ≤ i ≤ M, 1 ≤ j ≤ R. Ключевое наблюдение состоит в том, что во всех экземплярах Map используется одна и та же хэш-функция. Поэтому все эти экземпляры сохранят все выходные записи с одним и тем же значение хэш-функции в результирующем файле с одним и тем же номером.

На второй фазе MR-программа выполняет R экземпляров функции Reduce, где R обычно – это число узлов. Входные данные каждого экземпляра Reduce Rj состоят из файлов Fij, 1 ≤ i ≤ M. Эти файлы передаются по сети с локальных дисков узлов Map. Снова заметим, что все выходные записи фазы Map с одним и тем же значением хэш-функции попадают в один и тот же экземпляр Reduce независимо от того, какой экземпляр Map произвел эти данные. Каждый экземпляр Reduce обрабатывает или комбинирует назначенные ему записи и затем пишет записи в выходной файл (в распределенной файловой системе), образующий часть окончательного вывода данного вычисления.

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

Результаты и обсуждение

Результаты загрузки для наборов данных в 535 мегабайт на узел и 1 терабайт на кластер показаны на рис. 1 и 2 соответственно. Для СУБД-X разделено время двух фаз загрузки, что показано на диаграммах в виде составного прямоугольника: нижняя часть представляет время выполнения параллельной команды LOAD, а верхняя соответствует процессу реорганизации.

Рис. 1. Время загрузки – набор данных задачи Grep (535 мегабайт на узел)

Рис. 2. Время загрузки – набор данных задачи Grep (1 терабайт на кластер)

На рис. 1 наиболее удивительна разница во времени загрузки данных в СУБД-X по сравнению с Hadoop и Vertica. Несмотря на параллельное выполнение команды LOAD в каждом узле кластера на первой фазе процесса загрузки, в действительности, данные в каждом узле загружались последовательно. Поэтому при возрастании общего объема данных пропорционально увеличивается и время загрузки. Это также объясняет то, что для набора данных в один терабайт на кластер время загрузки для СУБД-X не уменьшается при сокращении объема данных в расчете на узел. Однако выполнение сжатия и других служебных операций на данными в СУБД-X может производиться параллельно в нескольких узлах, и поэтому время выполнения второй фазы процесса загрузки сокращается вдвое при увеличении вдвое числа узлов, используемых для хранения терабайта данных. При отсутствии использования сжатия на уровне блоков или записей Hadoop явным образом превосходит и СУБД-X, и Vertica, поскольку в каждом узле происходит всего лишь копирование файла данных с локального диска в локальный экземпляр HDFS, а затем две его копии передаются в другие узлы кластера. Если бы данные загружались в Hadoop с использованием только одной копии каждого блока, то время загрузки уменьшилось бы втрое. Но, как будет показано в разд. 5, отсутствие нескольких копий часто приводит к увеличению времени выполнения заданий.

Производительность всех трех систем при выполнении этой задачи показана на рис. 4 и 5. Удивительно, что на этих рисунках не согласуются относительные различия между системами. На рис. 4 две параллельные системы баз данных показывают почти одинаковую производительность, более чем в два раза превосходящую производительность Hadoop. Но на рис. 5 и СУБД-X, и Hadoop работают более чем в два раза медленнее Vertica. Причина состоит в том, что в этих двух экспериментах использовались данные существенно разного объема. Результаты, показанные на рис. 4, были получены на данных очень малого объема (535 мегабайт на узел). Это приводит к тому, что не столь уж незначительные накладные расходы на запуск Hadoop становятся фактором, ограничивающим производительность системы. Как будет показано в п. 5.1.2, для запросов с небольшим временем выполнения (меньше одной минуты) время запуска доминирует в общем времени выполнения. По наблюдениям авторов, проходит 10-25 секунд, пока все задачи Map стартуют и начинают выполняться с полной скоростью в узлах кластера. Кроме того, по мере увеличения общего числа этих задач появляются дополнительные накладные расходы для координации активности узлов центральным контроллером заданий. Эти накладные расходы незначительно возрастают при добавлении узлов к кластеру и, как показывает рис. 5, при выполнении более протяженных задач обработки данных затмеваются временем, затрачиваемым на требуемую обработку.

На приведенных диаграммах верхняя часть прямоугольников Hadoop представляет время выполнения дополнительного задания MR, объединяющего выходные данные в один файл. Поскольку эта заключительная фаза выполнялась в виде отдельного задания MapReduce, в случае, показанном на рис. 4, на нее тратилась большая часть общего времени выполнения задачи, чем в случае с рис. 5, т.к. накладные расходы на запуск снова доминировали над полезными затратами, требуемыми для выполнения завершающей фазы. Хотя задача Grep является селективной, результаты, показанные на рис. 5, демонстрируют, что для выполнения этой объединительной фазы могут потребоваться сотни секунд из-за потребности в открытии и объединении большого числа мелких выводных файлов. Каждый экземпляр Map помещает свои выводные данные в отдельный файл HDFS, и поэтому, хотя каждый файл невелик, имеется много задач Map и, следовательно, много файлов в каждом узле.

На рис. 5, на котором показаны результаты экспериментов с набором данных в один терабайт на кластер, видно, что все системы при увеличении вдвое числа узлов выполняют задачу почти вдвое быстрее. Этого и следовало ожидать, поскольку в этих экспериментах общий объем данных, распределенных по узлам, остается неизменным. Hadoop и СУБД-X демонстрируют примерно одинаковую производительность, поскольку в этих экспериментах накладные расходы на запуск Hadoop амортизируются возросшим объемом обрабатываемых данных. Однако результаты отчетливо показывают превосходство Vertica над СУБД-X и Hadoop. Авторы объясняют это активным использованием в Vertica сжатия данных (см. п. 5.1.3), которое становится более эффективным при хранении в каждом узле большего объема данных.

Как уже говорилось по поводу задачи Grep, результаты этих экспериментов, показанные на рис. 6, снова демонстрируют, что параллельные СУБД значительно превосходят Hadoop на всех уровнях масштабирования кластера. Хотя относительная производительность всех систем деградирует при возрастании числа узлов и общего объема данных, на Hadoop это действует сильнее всего. Например, время выполнения в экспериментах с одним узлом и 10 узлами различается почти на 50%. Это опять объясняется возрастающими накладными расходами на запуск системы при добавлении узлов к кластеру. При выполнении быстро обрабатываемых запросов эти накладные расходы занимают значительную часть общего времени выполнения.

Еще одной важной причиной, по которой параллельные системы могут превосходить Hadoop, является то, что и в Vertica, и в СУБД-X используется индекс на столбце pageRank, и таблица Rankings сохраняется уже отсортированной по значениям pageRank. Поэтому выполнение этого запроса тривиально. Следует также заметить, что хотя у Vertica абсолютное время выполнения задачи остается небольшим, относительная производительность системы деградирует при увеличении числа узлов. И это несмотря на тот факт, что в каждом узле запрос продолжает выполняться одно и то же время (около 170 микросекунд). Но, поскольку узлы завершают выполнение запроса настолько быстро, система становится заполненной управляющими сообщениями, передаваемыми из слишком многих узлов, и обработка этих сообщений занимает большее время. В Vertica используется надежный механизм передачи сообщений для распределения запроса по узлам и обработки протокола фиксации , и этот механизм порождает значительные накладные расходы при использовании для обработки запросов более нескольких десятков узлов.

результаты экспериментов с задачей агрегации, представленные на рис. 7 и 8, снова показывают превосходство двух СУБД над Hadoop. СУБД выполняют эти запросы путем сканирования в каждом узле соответствующей локальной таблицы, извлечения значений полей sourceIP и adRevenue и выполнения локальной группировки. Затем эти локальные группы объединяются координатором запроса, который выводит результаты пользователю. Результаты на рис. 7 демонстрируют, что при большом числе групп две СУБД показывают примерно одинаковую производительность, поскольку значительная часть времени тратится на передачу большого числа локальных групп и их объединение координатором. В экспериментах с меньшим числом узлов Vertica работает немного лучше, поскольку в ней читается меньше данных (имеется прямой доступ к столбцам sourceIP и adRevenue), но при увеличении числа узлов система слегка замедляется.

Результаты на рис. 8 показывают, что поколоночную систему выгоднее использовать при обработке для решения данной задачи меньшего числа групп. Это объясняется тем, что значения двух требуемых столбцов (sourceIP и adRevenue) состоят всего из 20 байт, а весь кортеж таблицы UserVisits занимает более 200 байт. Поэтому при наличии относительно небольшого числа групп, которые требуется объединять, коммуникационные расходы существенно ниже, чем при выполнении первого варианта запроса. Таким образом, Vertica обгоняет по производительности две другие системы за счет того, что не читает неиспользуемые части кортежей UserVisits. Заметим, что время выполнения запроса на всех системах почти одно и тоже для любого числа узлов (с учетом того, что Vertica при росте числа узлов слегка замедляется). Поскольку в этой задаче требуется, чтобы система просматривала весь набор данных, время выполнения всегда определяется эффективностью последовательного сканирования и сетевыми накладными расходами каждого узла.

Производительность систем при выполнении этой задачи демонстрируется на рис. 9. Из-за ошибки в оптимизаторе запросов авторам пришлось немного изменить код SQL, использовавшийся в эксперименте со 100 узлами для Vertica. Из-за этого наблюдается такое значительное увеличение времени выполнения при переходе от 50 к ста узлам. Но даже при этом очевидно, что именно на этой задаче наблюдается наибольшее различие в производительности между Hadoop и двумя параллельными системами баз данных. Причина этого различия двояка.

Во-первых, несмотря на возросшую сложность запроса, производительность Hadoop по-прежнему ограничена скоростью, с которой может быть прочитана с диска крупная таблица UserVisits (20 гигабайт на узел). MR-программа вынуждена производить полное сканирование таблиц, в то время как параллельные системы баз данных могли успешно воспользоваться кластеризованным индексом на столбце UserVisits.visitDate для значительного сокращения объема данных, которые требовалось прочитать. Сравнивая временные расходы Hadoop на выполнение разных фаз программы, авторы обнаружили, что, независимо от числа узлов в кластере, на выполнение фаз 2 и 3 тратилось в среднем 24.3 и 12.7 секунды соответственно. В отличие от этого, на выполнение фазы 1, содержащей задачу Map, которая читает таблицы UserVisits и Rankings, в среднем тратилось 1434.7 секунды. Интересно, что примерно 600 секунд из этого времени было затрачено на примитивное чтение с диска таблиц UserVisits и Rankings, а 300 секунд – на разбиение, разбор и десериализацию различных атрибутов. Таким образом, накладные расходы ЦП, требуемые для разбора таблиц на лету, являются для Hadoop ограничивающим фактором.

Во-вторых, параллельные СУБД могут опираться на тот факт, что обе таблицы, UserVisits и Rankings, разделяются по столбцу соединения. Это означает, что обе системы могут производить соединение в каждом узле локально, без сетевых расходов на повторное разделение до выполнения соединения. Таким образом, им нужно просто выполнить в каждом узле локальную операцию хэш-соединения таблицы Rankings и отфильтрованной части таблицы UserVisits с последующим тривиальным выполнением раздела ORDER BY.

Результаты на рис. 10 показывают, что и у СУБД-X, и у Hadoop (не включая дополнительный процесс Reduce для объединения данных) для этой задачи обеспечивается почти константная производительность, поскольку в каждом узле хранится один и тот же объем данных Document, предназначенных для обработки, и этот объем данных остается константой (7 гигабайт) при добавлении узлов в проводившихся экспериментах. Как и ожидалось, дополнительная операция Hadoop по объединению данных работает все медленнее при добавлении узлов, поскольку объем выходных данных, которые должны обрабатываться одним узлом, увеличивается. Результаты для СУБД-X и Vertica на рис. 10 показаны в виде составных прямоугольников, в которых нижняя часть представляет время, затрачиваемое на выполнение UDF/парсера и загрузку данных в таблицу, а верхняя часть показывает время, затраченное на выполнение реального запроса. СУБД-X демонстрирует производительность хуже, чем у Hadoop, из-за дополнительных накладных расходов на покортежное взаимодействие между UDF и входным файлом, хранимым вне баз данных. Плохая производительность Vertica является следствием потребности в разборе данных вне СУБД и материализации промежуточных результатов на локальном диске до их загрузки в систему.

Сравнение подходов к крупномасштабному анализу данных

Эндрю Павло, Эрик Паулсон, Александр Разин, Дэниэль Абади, Дэвид Девитт, Сэмюэль Мэдден, Майкл Стоунбрейкер
Пересказ: Сергей Кузнецов
Оригинал: Andrew Pavlo, Erik Paulson, Alexander Rasin, Daniel J. Abadi, David J. DeWitt, Samuel Madden, Michael Stonebraker. A Comparison of Approaches to Large-Scale Data Analysis. Proceedings of the 35th SIGMOD International Conference on Management of Data, 2009, Providence, Rhode Island, USA

Стратегии исполнения

Как отмечалось ранее, планировщик запросов в параллельных СУБД тщательно следит за тем, чтобы данные пересылались между узлами только в случае абсолютной необходимости. Это позволяет системам оптимизировать алгоритмы соединений в зависимости от характеристик данных и выполнять «проталкиваемую» передачу сообщений без потребности сохранения промежуточных наборов данных. Со временем сторонникам MR стоит изучить методы, используемые в параллельных СУБД, и перенять понятия, уместные для их модели. Авторы статьи полагают, что это позволит значительно повысить эффективность сред MR.

Кроме того, параллельные СУБД конструируют полный план выполнения запроса, который рассылается во все обрабатывающие узла в начале выполнения запроса. Поскольку данные «проталкиваются» между узлами только при необходимости, в течение обработки отсутствуют управляющие сообщения. В отличие от этого, в системах MR используется большое число управляющих сообщений для синхронизации обработки, что ухудшает производительность из-за возрастающих накладных расходов. СУБД Vertica также свойственна эта проблема, но в гораздо меньшем масштабе (см. подраздел 4.2).

Стратегия выполнения

Потенциально серьезная проблема производительности MR связана с управлением передачей данных от заданий Map к заданиям Reduce. Напомним, что каждый из N экземпляров MAP производит M выходных файлов, каждый из которых предназначен для соответствующего отдельного экземпляра Reduce. Эти файлы записываются на локальные диски в узлах, в которых выполняется каждый экземпляр Map. Если N = 1000, а M = 500, то на фазе Map данной программы будет произведено 500000 локальных файлов. Когда начинается фаза Reduce, каждому из 500 экземпляров Reduce требуется прочитать свою тысячу входных файлов, и при этом необходимо использовать протокол передачи файлов для «вытаскивания» (pull) каждого из своих входных файлов из узлов, на которых выполнялись экземпляры Map. При наличии сотен одновременно выполняющихся экземпляров Reduce неизбежно два или большее число этих экземпляров будут пытаться одновременно прочитать свои входные файлы с одного и того же узла Map, что приведет к большому числу подводов головок и замедлит скорость чтения данных с диска. Именно поэтому в параллельных системах баз данных разделенные файлы не материализуются, и вместо подхода «вытаскивания» для передачи данных используется подход «проталкивания» (push).

СУБД-X

Использовался последний выпуск СУБД-X – параллельной SQL-ориентированной СУБД одного из ведущих поставщиков таких систем, в которой данные хранятся в «построчном» формате. Эта система инсталлировалась в каждом узле и конфигурировалась таким образом, что для буферного пула и других временных областей использовались сегменты разделяемой памяти общим объемом в 4 гигабайта. Каждая таблица хэш-разделяется по всем узлам в соответствии со значениями своего наиболее существенного атрибута, а затем сортируется и индексируется по разным атрибутам (см. пп. 4.2.1 и 4.3.1). Как и в экспериментах с Hadoop, перед каждым следующим прогоном в СУБД-X удалялись все таблицы, и данные загружались заново, чтобы кортежи были гарантированно равномерно распределены по узлам кластера.

По умолчанию СУБД-X не сжимает данные в своей внутренней системе хранения, но обеспечивается возможность сжатия таблиц с использованием широко известной схемы, основанной на словарях. Было установлено, что при включении опции сжатия время выполнения почти всех тестов сокращается на 50%, и поэтому ниже приводятся только те результаты, которые были получены при использовании сжатия данных. Только в одном случае оказалось, что при использовании сжатия производительность системы ухудшается. Кроме того, в СУБД-X не использовались средства репликации, поскольку это не улучшило бы производительность системы и усложнило бы процесс инсталляции.

Процесс загрузки в СУБД-X происходил в два этапа. Сначала в каждом узле кластера в параллель выполнялась команда SQL LOAD для чтения данных из локальной файловой системы, и их содержимое вставлялось в соответствующую таблицу базы данных. В этой команде указывалось, что локальные данные разделены некоторым специальным символом, так что не требовалось писать специальную программу для преобразования данных до их загрузки. Но, поскольку генератор тестовых данных просто создавал случайные ключи для каждой записи в каждом узле, системе было необходимо перераспределить кортежи по другим узлам кластера на основе атрибута разделения целевой таблицы. Можно было бы создать вариант генератора данных, учитывающий хэширование, что позволило бы СУБД-X просто загрузить вводные файлы в каждом узле без этого процесса перераспределения, но вряд ли это слишком сократило бы общее время загрузки.

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

Сжатие

Почти во всех параллельных СУБД (включая СУБД-X Vertica) допускается опциональное сжатие хранимых данных. Нередко при сжатии данных удается сократить требуемый объем памяти в 6-10 раз. Внутреннее представление данных в Vertica в высшей степени оптимизировано для сжатия данных, и обработчик запросов работает напрямую со сжатыми данными (т.е. по мере возможности избегает распаковки данных при их обработке). Как правило, поскольку аналитические задачи над большими наборами данных часто тормозятся вводом-выводом, жертвование циклами ЦП (требуемыми для распаковки входных данных) в пользу пропускной способности ввода-вывода (сжатие данных позволяет считывать меньше данных с диска) является правильной стратегией и приводит к более быстрому выполнению задач. В ситуациях, в которых обработчик запросов может оперировать прямо со сжатыми данными, часто вообще не требуются никакие компромиссы, и сжатие, очевидно, всегда приносит пользу.

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

Чтобы использовать в Hadoop сжатие на уровне блоков, сначала требуется разбить файлы данных на несколько более мелких файлов в локальной файловой системе каждого узла, а затем сжать каждый файл с использованием инструмента gzip. Сжатие данных таким способом сокращает каждый набор данных на 20-25% от его исходного размера. Затем эти сжатые файлы копируются в HDFS ровно так же, как если бы они были плоскими текстовыми файлами. Hadoop автоматически определяет, когда файлы являются сжатыми, и автоматически распаковывает их «на лету», когда они передаются экземплярам Map, так что для использования сжатых данных не приходится изменять MR-программы. Несмотря на большее время загрузки (если включать в него время на разбиение и сжатие файлов), Hadoop, использующий сжатие на уровне блоков, замедляет выполнение большинства задач на несколько секунд, а задачи, для которых требуются в основном ресурсы процессора, выполнялись на 50% дольше.

Авторы также пытались выполнять тестовые задачи с использованием сжатия на уровне записей. Для этого требовалось (1) написать собственный класс tuple с использованием Hadoop API, (2) модифицировать программу загрузки данных, чтобы она сжимала записи и сериализовала получаемые кортежи, и (3) произвести рефакторинг всех тестовых задач. Вначале имелась надежда, что это позволит повысить производительность задач с высокими требованиями к ресурсам ЦП, поскольку задачам Map и Reduce больше не требовалось расщеплять поля по разделителю. Однако было установлено, что этот подход обеспечивает производительность, еще более низкую, чем сжатие на уровне блоков, сжимая данные только на 10%.

Тестовая среда

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

Тесты для оценки производительности

В этом разделе описывается тестовый набор, состоящий из пяти задач, используемых для сравнения производительности модели MR и параллельных СУБД. Первая задача была взята прямо из исходной статьи про MapReduce , авторы которой объявляют ее типичной для MR. Поскольку эта задача довольно проста, были разработаны четыре дополнительных задачи, которые образуют более сложную аналитическую нагрузку, позволяющую исследовать архитектурные различия, обсуждавшиеся в предыдущем разделе. Тестовый набор выполнялся на широко известной реализации MR и на двух параллельных СУБД.

Vertica

Vertica – это параллельная СУБД, разработанная для управления крупными хранилищами данных . Основным отличием Vertica от других СУБД (включая СУБД-X) является то, что эта система хранит все данные по столбцам, а не по строкам . В системе используется уникальный обработчик запросов, специально разработанный для применения в среде поколоночного хранения данных. В отличие от СУБД-X, в Vertica данные сжимаются по умолчанию, поскольку исполнитель запросов может работать прямо со сжатыми данными. Поскольку при реальном применении Vertica не принято запрещать возможность сжатия данных, при получении результатов, представленных в этой статье, использовались только сжатые данные. Кроме того, в Vertica все таблицы сортируются по значениям одного или нескольких атрибутов на основе кластеризованного индекса.

Было обнаружено, что в проводившихся экспериментах вполне был достаточен буфер с размером по умолчанию в 256 мегабайт в расчете на один узел. Менеджер ресурсов Vertica отвечает за определение объема памяти, выделяемой для выполнения каждого запроса, но в данном случае система была оповещена о том, что запросы следует выполнять по одному. Поэтому на каждом узле для выполнения каждого запроса использовался максимально доступный объем памяти.

В Vertica также поддерживается команда SQL COPY, которая задается из одного узла, а затем координирует процесс загрузки, выполняемый параллельно в нескольких узлах кластера. В качестве входных данных команде COPY задается список узлов, для которых требуется выполнить операцию загрузки. Этот процесс аналогичен тому, который выполняется для СУБД-X: в каждом узле загрузчик системы Vertica расщепляет файлы исходных данных по некоторому разделителю, создает новый кортеж для каждой строки вводного файла и помещает этот кортеж в соответствующий узел на основе значения хэш-функции от его первичного ключа. После загрузки данных столбцы автоматически сортируются и сжимаются в соответствии с физической схемой базы данных.

Аналогично СУБД-X, в Vertica использовались те же команды массовой загрузки, что обсуждались в подразделе 4.2, и таблицы UserVisits и Rankings сортировались по столбцам visitDate и pageRank соответственно.

Рис. 3. Время загрузки – набор данных UserVisits (20 гигабайт на узел)

В последнее время специализированные издания

В последнее время специализированные издания переполнены новостями о революции «кластерных вычислений». Эта парадигма состоит в использовании большого числа (низкопроизводительных) процессоров, работающих в параллель для решения вычислительной проблемы. По существу, предлагается построение центра данных путем объединения большого числа низкопроизводительных серверов вместо использования меньшего числа высокопроизводительных серверов. Рост интереса к кластерам способствует распространению средств их программирования. Одним из первых и наиболее известных подобных средств является MapReduce (MR) . Подход MapReduce привлекателен тем, что обеспечивает простую модель, на основе которой пользователи могут выражать сравнительно сложные распределенные программы, что порождает значительные интерес в образовательном сообществе. Например, IBM и Google обнародовали планы по обеспечению доступности 1000-процессорного кластера MapReduce для обучения студентов распределенному программированию.
При наличии этого интереса к MapReduce естественно задать вопрос: «А почему бы вместо этого не использовать какую-нибудь параллельную СУБД?». Параллельные системы баз данных (которые все основаны на общих архитектурных принципах) коммерчески доступны уже почти двадцать лет, и на рынке их около десятка, включая Teradata, Aster Data, Netezza, DATAllegro (и, следовательно, вскоре Microsoft SQL Server через посредство проекта Madison), Dataupia, Vertica, ParAccel, Neoview, Greenplum, DB2 (посредством Database Partitioning Feature) и Oracle (посредством Exadata). Это надежные, высокопроизводительные вычислительные платформы. Подобно MapReduce, они обеспечивают среду высокоуровневого программирования и полную распараллеливаемость. Хотя может показаться, что MR и параллельные системы баз данных нацелены на разную публику, на самом деле, можно написать почти любую задачу параллельной обработки либо в виде некоторого набора запросов к базе данных (возможно, с использованием определяемых пользователями функций и агрегатов для фильтрации и комбинирования данных), либо в виде набора заданий MapReduce.


Вдохновляемые этим вопросом, авторы задались целью понять, в чем состоят различия подходов MapReduce и параллельных систем баз данных при выполнении крупномасштабного анализа данных. Эти два класса систем расходятся в нескольких ключевых аспектах. Например, для всех СУБД требуются данные, соответствующие строго определенной схеме, в то время как MR допускает использование данных, представленных в любом произвольном формате. К числу других отличий относятся способы оптимизации на основе индексации и сжатия данных, модели программирования, методы распределения данных и стратегии выполнения запросов.
Цель статьи состоит в том, чтобы проанализировать эти отличия и их последствия. Второй раздел статьи начинается с краткого обзора этих двух альтернативных классов систем, после чего в разделе 3 обсуждается их архитектурные особенности. Затем в разделе 4 описывается эталонный тестовый набор, состоящий из разнообразных задач, одна из которых взята из статьи про MR , а прочие являются более трудными. Кроме того, приводятся результаты прогонов этого тестового набора на 100-узловом кластере. В испытаниях участвовали публично доступная версия MapReduce с открытыми кодами Hadoop , а также две параллельных SQL-ориентированных СУБД – Vertica и система одного из основных поставщиков реляционных СУБД. Также приводятся данные о временных затратах на загрузку и проверку данных, и неформально описываются процедуры, потребовавшиеся для установки и настройки программного обеспечения для каждой задачи.
В большинстве случаев SQL-ориентированные СУБД оказались существенно более быстрыми, и при их использовании потребовалось меньше кода для реализации каждой задачи, но больше времени для настройки и загрузки данных. На основе полученных результатов в заключении статьи обсуждаются причины различий между рассматриваемыми подходами, и приводятся рекомендации по поводу оптимальных методов для любого средства крупномасштабного анализа данных.
Некоторые читатели могут счесть, что эксперименты, проводимые с использованием 100 узлов, не являются интересными или представительными с точки зрения реальных систем обработки данных.


Авторы не согласны с этим предположением в двух отношениях. Во-первых, как демонстрируется в разд. 4, на 100 узлах две параллельные СУБД справляются с разными аналитическими задачами в 3,1-6,5 раз быстрее, чем MapReduce. Хотя, конечно, MR может масштабироваться до тысяч узлов, из-за исключительной эффективности современных СУБД такая массивная аппаратура не требуется даже при наличии наборов данных в 1-2 петабайта (1000 узлов с двухтерабайтной дисковой памятью на узел обладают общей дисковой емкостью в 2 петабайта). Например, в конфигурации Teradata в eBay используются всего 72 узла (в каждом узле два четырехъядерных процессора, 32 гигабайта основной памяти и 104 300-гигабайтных диска) для управления реляционными данными объемом около 2,4 петабайт. В качестве другого примера, хранилище данных Fox Interactive Media реализуется с использованием СУБД Greenplum на 40 узлах. Каждый узел представляет собой машину Sun X4500 с двумя двухъядерными процессорами, дисками общей емкостью в 48500 гигабайт и 16 гигабайтами основной памяти (1 петабайт общей дисковой памяти) . Поскольку петабайтного размера в мире достигают лишь немногие наборы данных, совсем непонятно, скольким пользователям MR на самом деле требуется 1000 узлов.

Выполнение тестов

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

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

Во всех случаях, кроме специально оговариваемых, окончательные результаты запросов, выполнявшихся на Vertica или СУБД-X, направлялись командой shell по программному каналу (pipe) в файл на диске, который не использовался СУБД. Хотя можно выполнить эквивалентную операцию и в Hadoop, проще (и более распространено) сохранять результаты MR-программы в распределенной файловой системе. Однако эта процедура не аналогична тому, как производят свои выводные данные СУБД. Вместо того чтобы сохранять результаты в выводном файле, MR-программа производит один выводной файл для каждого узла Reduce и сохраняет все эти файлы в одном каталоге. Стандартная практика разработчиков состоит в использовании этих выводных каталогов как единиц ввода для других MR-заданий. Однако если некоторый пользователь пожелает использовать эти выводные каталоги в не MR-приложениях, им приходится сначала объединить результаты в одном файле и загрузить его в локальную файловую систему.

Из-за этого различия для каждой тестовой MR-задачи выполнялась дополнительная функция Reduce, которая просто объединяла окончательный вывод в одном файле HDFS. В приводимых результатах отдельно приводится время, затраченное Hadoop на выполнение реальной задачи тестового набора, и время выполнения дополнительной операции объединения выводных данных. Эти результаты демонстрируются в виде составных прямоугольников: в нижней части показано время выполнения конкретной тестовой задачи, а в верхней – время выполнения функции Reduce, объединяющей в один файл все выводные данные программы.

Задача Aggregation

В следующей задаче требуется, чтобы каждая из систем вычислила суммарное значение adRevenue для каждой группы кортежей таблицы UserVisits (20 гигабайт на узел) с одним и тем же значением столбца sourceIP. Кроме того, запускался вариант этого запроса, в котором группировка производилась по первым семи символам значений столбца sourceIP, чтобы выяснить, как влияет на эффективность выполнения запроса сокращение числа групп. Эта задача была разработана для определения эффективности параллельной аналитики над единственной только читаемой таблицей. В этом случае для вычисления окончательного результата узлам требуется обмениваться промежуточными данными. Независимо от числа узлов в кластере эта задача всегда производит 2.5 миллиона записей (53 мегабайта); при выполнении варианта запроса с меньшим числом групп производится 2000 записей (24 килобайта).

Задача Join

Задача соединения состоит из двух подзадач, выполняющих сложные вычисления над двумя наборами данных. В первой части задачи каждая система должна найти sourceIP, которые принесли наибольшую выручку в заданном интервале времени. После образования этих промежуточных записей система должна вычислить среднее значение pageRank для всех страниц, посещенных в течение этого интервала. В экспериментах использовался интервал от 15 до 22 января 2000 г., которому соответствует примерно 134000 записей в таблице UserVisits.

Основной особенностью этой задачи является то, что она должна использовать два разных набора данных и соединить их для нахождения пар записей Ranking и UserVisits, у которых совпадают значения pageURL и destURL. Для решения этой задачи в каждой системе приходится использовать достаточно сложные операции над данными большого объема. Результаты эффективности также позволяют установить, насколько хорошо оптимизаторы запросов СУБД производят эффективные планы выполнения запросов.

Задача Selection

Задача Selection – это легковесный фильтр, предназначенный для нахождения значений pageURL в таблице Rankings (1 гигабайт на узел), для которых значение pageRank превышает заданное пользователем пороговое значение. В описываемых экспериментах в качестве значения порогового параметра использовалось 10, что приводило к выборке примерно 36000 записей в каждом узле.

Задача UDF Aggregation

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

Авторы вносят в постановку задачи две корректировки с целью облегчить ее выполнение в Hadoop. Во-первых, в агрегате допускается учет ссылок из документа на самого себя, поскольку в функции Map нетривиально обнаружить имя обрабатываемого файла. Во-вторых, в каждом узле HTML-документы конкатенируются в более крупные файлы при их сохранении в HDFS. Авторы обнаружили, что это позволяет повысить производительность Hadoop в два раза и помогает избежать проблем с основной памятью при использовании центрального контроллера HDFS, когда в системе сохраняется большое число файлов.

Загрузка данных

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

Опишем процедуры загрузки наборов данных UserVisits и Rankings. По соображениям, обсуждаемым в п. 4.3.5, только для Hadoop требовалось непосредственно загружать файлы Documents во внутреннюю систему хранения. И в СУБД-X, и в Vertica выполнялась UDF, которая обрабатывала Documents в каждом узле во время выполнения и загружала данные во временную таблицу. Накладные расходы этого подхода учитываются во времени прогона тестов, а не во времени загрузки данных. Поэтому результаты загрузки этого набора данных в статье не приводятся.

Загрузка и размещение данных

В параллельных СУБД имеется возможность реорганизации входных данных во время загрузки. Это позволяет производить некоторые оптимизации, такие как хранение каждого атрибута таблиц по отдельности (как это делается в поколоночных системах типа Vertica). Для запросов на выборку данных, которые затрагивают только часть атрибутов таблицы, эта оптимизация может существенно повысить производительность, позволяя не читать с диска значения атрибутов, которые не требуются для выполнения данного запроса. Аналогично описанной выше оптимизации сжатия данных, это позволяет более рационально использовать пропускную способность ввода-вывода. Системы MR по умолчанию не преобразуют данные при их загрузке в распределенную файловую систему, и поэтому они не способны изменить схему хранения входных данных, что препятствует выполнению оптимизаций отмеченного класса. Кроме того, для Hadoop всегда требовалось больше ресурсов ЦП, чем для параллельных СУБД, при пропуске эквивалентных задач, поскольку этой системе приходится разбирать и десериализовывать записи входных данных во время выполнения, в то время как параллельные системы баз данных производят разбор во время загрузки данных и могут быстро извлекать значения атрибутов кортежей практически без накладных расходов.

Но упрощенный процесс загрузки в MR гораздо проще и быстрее загрузки данных в СУБД. Результаты п. 4.2.1 и 4.3.1 показывают, что загрузка данных в Hadoop происходит более чем в три раза быстрее, чем в Vertica, и почти в 20 раз быстрее, чем в СУБД-X. Это говорит о том, что для данных, которые загружаются только для решения некоторых типов аналитических задач, может оказаться нецелесообразно тратить дополнительные расходы на их индексацию и реорганизацию, свойственные СУБД. Это также говорит о том, что СУБД могут выиграть от поддержки режима обработки данных «на месте», позволяющего пользователям непосредственно обращаться и направлять запросы к файлам, сохраняемым в локальной файловой системе.

в этой статье, можно сделать

На основе результатов, представленных в этой статье, можно сделать ряд интересных выводов. Во-первых, в масштабе выполненных экспериментов обе параллельные системы продемонстрировали существенное превосходство в производительности над Hadoop при выполнении ряда тестовых задач анализа данных большого объема. В среднем для всех пяти задач на кластере из 100 узлов СУБД-X оказалась в 3.2 раза быстрее MR, а Vertica – в 2.3 раза быстрее СУБД-X. Хотя авторы не могут проверить это утверждение, они полагают, что те же относительные показатели производительности сохранились бы и на кластере с 1000 узлов (самая крупная конфигурация Teradata на кластере с менее чем 100 узлами управляет данными объемом более четырех петабайт). Эти цифры можно истолковать так, что параллельные системы баз данных, обеспечивающие то же самое время отклика при использовании меньшего числа процессоров, безусловно, будут потреблять намного меньше энергии; модель MapReduce на кластере с несколькими тысячами узлов является решением грубой силы, растрачивающим огромные объемы энергии. Хотя ходят слухи, что версия MR от Google быстрее версии Hadoop, у авторов статьи не было доступа к этому коду, и поэтому они не могли его протестировать. Однако они сомневаются, что эти две версии MR показали бы существенную разницу в производительности, поскольку MR вынуждает всегда начинать выполнение запроса с просмотра всего входного файла.
Это преимущество в производительности, свойственное обеим системам баз данных, является результатом применения ряда технологий, разработанных в последние 25 лет, включая (1) индексы в виде B-деревьев, ускоряющие выполнение операций выборки, (2) новые механизмы хранения данных (например, поколоночное хранение), (3) эффективные методы сжатия данных с возможностью выполнять операции прямо над сжатыми данными, (4) сложные параллельные алгоритмы для выполнения запросов над большими объемами реляционных данных. В случае применения поколоночных систем баз данных, подобных Vertica, с диска считываются только те столбцы, которые требуются для выполнения запроса.
Кроме того, поколоночное хранение позволяет добиться лучших показателей сжатия данных (в Vertica они сжимаются примерно в 2.0 раза, в СУБД-X – в 1.8 раз, а в Hadoop – в 1.25 раз). Это тоже способствует сокращению объема дискового ввода-вывода при выполнении запросов.
Хотя авторы не были удивлены относительным превосходством в производительности двух параллельных систем баз данных, на них произвела впечатление простота установки Hadoop и использования этой системы по сравнению с системами баз данных. Процесс инсталляции Vertica тоже был несложным, но чувствительным к некоторым системным параметрам. С другой стороны, СУБД-X было трудно сконфигурировать должным образом, и потребовалась неоднократная помощь от представителей компании-поставщика для получения правильно работающей конфигурации. В целом, установка и конфигурирование такого зрелого продукта, как СУБД-X, произвели на авторов неутешительное впечатление. Учитывая безусловное стоимостное преимущество Hadoop, можно понять, почему эта система так быстро привлекла к себе крупное сообщество пользователей.
Еще одной областью, в которой во время тестирования систем баз данных были обнаружены недостатки, является расширяемость. Идее расширения СУБД определяемыми пользователями типами и функциями уже 25 лет [16]. Тем не менее, ни одна из тестируемых параллельных систем не смогла хорошо справиться с задачей агрегации с применением UDF, заставив авторов искаться обходные пути при обнаружении ограничений (например, в Vertica) или ошибок (например, в СУБД-X).
Хотя системы баз данных устойчивы к разнообразным программным сбоям, нет никаких сомнений, что MR лучше справляется с минимизацией потерь уже выполненной работы при возникновении аппаратных сбоев. Однако эта возможность потенциально сопровождается крупным ухудшением производительности из-за расходов на материализацию промежуточных файлов между фазами Map и Reduce. Пока неясно, насколько существенно это ухудшение. К сожалению, для должного исследования этого вопроса требуется реализация в одной среде стратегий с материализацией и без нее, а такая работа не предусматривалась в исследованиях, которым посвящена данная статья.


Несмотря на очевидное преимущество Hadoop в этой области, не совсем ясно, насколько существенна устойчивость систем к аппаратным сбоям при их реальном практическом использовании. Кроме того, если для системы MR требуется 1000 узлов для достижения производительности параллельной системы баз данных, работающей на ста узлах, то для первой системы вероятность отказа узла при обработке запроса в десять раз больше, чем для второй. Тем не менее, от улучшенной устойчивости не отказался бы ни один пользователь баз данных.
Многие люди считают, что SQL поначалу трудно использовать. Частично это связано с тем, что при решении проблем с использованием SQL требуется несколько иной стиль мышления, и с тем, что SQL превратился в сложный язык, который существенно отличается от исходной разработки, выполненной Доном Чемберлином (Don Chamberlin) в 1970-х гг. Хотя большинство языков со временем усложняется, SQL особенно плох, поскольку многие его средства разрабатывались конкурирующими компаниями-поставщиками СУБД, каждая из которых стремилась включить в язык собственные проприетарные расширения.
Несмотря на свои недостатки, SQL по-прежнему является мощным инструментом. Рассмотрим запрос для генерации списка служащих, упорядоченного по заработной плате, причем для каждого служащего должен указываться еще и уровень его зарплаты (уровень зарплаты служащих, получающих максимальную зарплату, равен единице). На SQL этот запрос можно сформулировать следующим образом:
SELECT Emp.name, Emp.salary, RANK() OVER (ORDER BY Emp.salary) FROM Employees AS Emp
При параллельном выполнении этого запроса требуется полностью упорядочить всех служащих, после чего выполняется вторая фаза, на которой в каждом узле значения уровня для содержащихся в нем записей корректируются счетчиками числа записей со всех узлов «слева» от данного (т.е. тех узлов, в которых значения зарплаты строго меньше). Хотя MR-программа могла бы параллельно выполнить эту сортировку, не так просто подстроить этот запрос под парадигму MR группировки по агрегации.


RANK – это лишь одна из многих мощных аналитических функций, поддерживаемых в современных параллельных системах баз данных. Например, и в Teradata, и в Oracle поддерживается развитый набор функций, таких как функции над окнами упорядоченных записей.
Два архитектурных различия, похоже, сохранятся в течение длительного времени. MR следует парадигме «схема потом» или даже «вообще без схемы». Но это отсутствие схемы влечет ряд важных последствий. Прежде всего, это означает неизбежность разбора записей во время выполнения, в том время как СУБД производят разбор во время загрузки данных. Это различие делает менее полезным сжатие данных в среде MR и служит частичным источником различия в производительности между двумя классами систем. Во-вторых, схема требуется для поддержки информации, важной для оптимизации декларативных запросов, включая информацию о существующих индексах, разделении таблиц, мощности таблиц, а также гистограммы, представляющие распределения значений в столбцах.
По мнению авторов, для обоих видов систем можно еще многое сделать. Наиболее важны появление над базисным уровнем MR интерфейсов более высокого уровня, таких как Pig [15] и Hive [2], а также разработка инструментов, близких по духу к MR, но более выразительных, таких как Dryad [13] и Scope [5]. Это упростит кодирование сложных задач в MR-подобных системах и устранит одно из больших преимуществ SQL-ориентированных систем – меньшие трудозатраты на кодирование задач. Что касается параллельных систем баз данных, как коммерческих, так и с открытыми исходными текстами, авторы полагают, что в них будет существенно усовершенствован параллелизм функций, определяемых пользователями. Таким образом, API обеих классов систем, очевидно, сближаются. Первыми признаками этого являются решения по интеграции SQL и MR от компаний Greenplum и Asterdata.

Запуск задач

Было обнаружено, что при выполнении MR-программ проходит некоторое время до того, как все узлы начинают работать на полную мощность. На кластере из 100 узлов проходит 10 секунд от поступления задания в JobTracker до начала выполнения первой задачи Map и 25 секунд до того времени, когда это задание выполняется на всех узлах кластера. Это согласуется с результатами, описанными в , где пиковая скорость обработки данных достигалась в течении 60 секунд на кластере с 1800 узлами. «Холодный старт» характерен для реализации Hadoop (и, по-видимому, для соответствующей реализации Google), но не присущ самой модели MR. Например, авторы также обнаружили, что в предыдущих версиях Hadoop для каждого экземпляра Map и Reduce в соответствующем узле создавался новый процесс JVM, что повышало уровень накладных расходов при выполнении заданий над крупными наборами данных. Возможность повторного использования JVM в последней версии Hadoop повысила эффективность Hadoop при решении описываемых тестовых задач на 10-15%.

В отличие от этого, параллельные СУБД запускаются при раскрутке ОС, и поэтому их можно всегда считать «теплыми», ожидающими запросов для выполнения. Кроме того, все современные СУБД разработаны для выполнения запросов с использованием нескольких потоков управления и процессов, что позволяет уже исполняемому коду СУБД принимать дополнительные задачи и дополнительно оптимизировать план исполнения. Минимизация времени запуска была одной из первых оптимизаций СУБД, и, конечно, это то, что следует постараться внедрить в системы MR без серьезного изменения основной архитектуры.

---

---