Интеграция Hadoop и параллельной СУБД

Ю Ксу, Пекка Костамаа, Лайк Гао
Перевод: Сергей Кузнецов

Оригинал: Yu Xu, Pekka Kostamaa, Like Gao. Integrating Hadoop and Parallel DBMS. Proceedings of the 2010 International Conference on Management of Data (SIGMOD 2010), June 6-11, 2010, Indianapolis, Indiana, USA, pp. 969-974

This translation is a derivative of ACM-copyrighted material. ACM did not prepare this translation and does not guarantee that it is an accurate copy of the originally published work. The original intellectual property contained in this work remains the property of ACM.

Содержание

От переводчика: теперь и Teradata...

Аннотация

1. Введение

2. Параллельная загрузка данных Hadoop в Teradata EDW

3. Выборка данных EDW в программах MapReduce

3.1. DBInputFormat

3.2. TeradataInputFormat

4. Доступ к данным Hadoop из SQL с использованием табличной UDF

5. Родственные работы

6. Заключение

7. Литература

От переводчика: теперь и Teradata...

Честно скажу, я не в восторге от статьи, перевод которой вам предлагается. Она написана явно людьми "от сохи", технарями известнейшей компании, которые не балуют себя частым написанием исследовательских статей. Статья написана, мягко говоря, посредственно, в ней отсутствует описание экспериментов и т.д. Почему же я взялся за ее перевод?

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

Во-вторых, мною двигала и чисто коллекционерская цель. За 2009-2010 гг. годы я прочитал и перевел несколько хороших статей, посвященных скрещиванию технологий MapReduce и массивно-параллельных баз данных:

• Эндрю Павло, Эрик Паулсон, Александр Разин, Дэниэль Абади, Дэвид Девитт, Сэмюэль Мэдден, Майкл Стоунбрейкер. Сравнение подходов к крупномасштабному анализу данных
• Майкл Стоунбрейкер, Дэниэль Абади, Дэвит Девитт, Сэм Мэдден, Эрик Паулсон, Эндрю Павло и Александр Разин. MapReduce и параллельные СУБД: друзья или враги?
• Джеффри Коэн, Брайен Долэн, Марк Данлэп, Джозеф Хеллерстейн, Кейлэб Велтон. МОГучие способности: новые приемы анализа больших данных
• Эрик Фридман, Питер Павловски и Джон Кислевич. SQL/MapReduce: практический подход к поддержке самоописываемых, полиморфных и параллелизуемых функций, определяемых пользователями
• Азза Абузейд, Камил Байда-Павликовски, Дэниэль Абади, Ави Зильбершац, Александр Разин. HadoopDB: архитектурный гибрид технологий MapReduce и СУБД для аналитических рабочих нагрузок.

Я написал свою собственную обзорную статью MapReduce: внутри, снаружи или сбоку от параллельных СУБД?. Эта тема продолжает оставаться для меня очень интересной, и я стараюсь не пропускать статей, которые ее как-нибудь затрагивают. А статья Ю Ксу и др., конечно, этой темы непосредственно касается. Авторы идут по пути, близкому пути Vertica. Они не пытаются скрестить Teradata с MapReduce, засунув одно в другое (для Teradata такие потрясения вряд ли допустимы), а предлагают механизмы для плодотворного сосуществования: средства разного рода экспорта данных из среды Hadoop в среду Teradata и наоборот.

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

Аннотация

Параллельная СУБД Teradata на протяжении последних двадцати лет успешно используется в крупных хранилищах данных для выполнения крупномаштабного бизнес-анализа в различных областях индустрии над наборами данных объемом от нескольких терабайт до нескольких петабайт. Однако вследствии наблюдаемого в последние годы взрывообразного роста объема данных в вычислительных центрах некоторых заказчиков некоторые данные, такие как Web-журналы и сенсорные данные, не управляются Teradata EDW (Enterprise Data Warehouse), частично из-за того, что загрузка этих данных в РСУБД обходится слишком дорого, в особенности в тех случаях, когда эти данные не слишком часто используются для поддержки принятия важных бизнес-решений. В последнее время в академических и производственных кругах все более распространенным становится применение парадигмы MapReduce (придуманной в Google и ставшей популярной благодаря доступной в исходных кодах реализации Hadoop, основную поддержку которой оказывает Yahoo!) в качестве альтернативного способа выполнения крупномасштабного анализа данных. К настоящему времени большинство исследователей и практических специалистов в области хранилищ данных соглашается с тем, что у парадигм параллельных СУБД и MapReduce имеются свои достоинства и недостатки в разных бизнес-приложениях, и что поэтому этим двум парадигмам суждено сосуществовать на протяжении долгого времени [16]. На самом деле, большое число заказчиков Teradata, в особенности, те из них, кто относится к индустриям электронной коммерции и телекоммуникаций, испытывает возрастающую потребность в выполнении бизнес-анализа данных, сохраняемых и в Hadoop, и в Teradata EDW. Одной из общих черт Hadoop и Teradata EDW является то, что данные в обеих системах для параллельной обработки разделяются по нескольким узлам, что обеспечивает возможности интеграционной оптимизации, недоступные для СУБД, работающих в одном узле. В этой статье мы описываем три свои работы, направленные на достижение тесной и эффективной интеграции Hadoop и Teradata EDW.

1. Введение

Распределенные файловые системы (Distributed File System, DFS) широко используются в поисковых системах для хранения огромного объема данных, собираемых в Internet, поскольку DFS обеспечивают масштабируемое, надежное и экономичное решение хранения данных. Компании, специализирующиеся на разработке поисковых систем, также создают на основе DFS параллельные вычислительные платформы для параллельного выполнения крупномасштабного анализа данных, сохраняемых в DFS. Например, у Google имеются GFS [10] и MapReduce [8]. Yahoo! использует Hadoop [11] – реализацию с открытыми исходными текстами, выполненную Apache Software Foundation и основанную на GFS и MapReduce компании Google. Компания Ask.com построила Neptune [5]. У Microsoft имеются Dryad [13] и Scope [4].

Hadoop привлекает внимание большого сообщества пользователей по причине открытости кодов и наличия серьезной поддержки со стороны Yahoo!. В Hadoop файлы разбиваются на блоки, и каждый блок несколько раз реплицируется в разных узлах для обеспечения отказоустойчивости и распараллеливания вычислений. Обычно Hadoop выполняется в кластерах, построенных на основе недорогой аппаратуры массового спроса. Hadoop легко устанавливается, и системой просто управлять. Загрузка данных в DFS производится более эффективно, чем в параллельную СУБД [15].

Текущая тенденция состоит в том, что компании начинают использовать Hadoop для крупномасштабного анализа данных. Хотя для начала использования Hadoop требуются совсем небольшие расходы, обычно Hadoop MapReduce значительно уступает параллельным СУБД в производительности: Hadoop в 2-3 раза медленнее, чем параллельная СУБД, решает простейшую задачу подсчета числа вхождений разных слов в файле/таблице, и в десятки раз медленнее справляется с более сложными задачами анализа данных [15]. Кроме того, программы MapReduce для сложного анализа данных пишутся гораздо дольше, чем соответствующие SQL-запросы. Нам известно, что одна из крупных Internet-компаний, имеющая крупные кластеры с Hadoop, переходит к использованию параллельной СУБД для производства некоторых наиболее сложных аналитических отчетов, поскольку руководители компании не удовлетворены тем, что в обстановке постоянно изменяющихся и усложняющихся бизнес-требований им приходится ждать по несколько дней, пока будут написаны и отлажены требуемые сложные программы MapReduce.

С другой стороны, из-за того, что в вычислительных центрах некоторых заказчиков Teradata в последние годы наблюдается быстрый рост объемов данных, некоторые данные, такие как Web-журналы, детальные данные об обращениях клиентов, сенсорные данные и данные RFID не управляются Teradata EDW. Частично это связано с очень высокой стоимостью загрузки этих исключительно объемных данных в РСУБД, особенно, если эти данные не слишком часто используются для поддержки принятия важных бизнес-решений.

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

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

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

• Мы обеспечиваем утилиту полностью параллельной загрузки, называемую DirectLoad, для эффективной загрузки данных Hadoop в Teradata EDW. Ключевая идея подхода DirectLoad состоит в том, что сначала мы приписываем каждый блок данных файла Hadoop некоторому параллельно компоненту Teradata EDW, а затем напрямую параллельно загружаем данные в параллельные компоненты. Для поддержки подхода Teradata EDW мы также применяем внутри Teradata EDW новые методы для минимизации перемещения данных между узлами.
• Мы обеспечиваем коннектор для Hadoop под названием TeradataInputFormat, который позволяет программам MapReduce напрямую читать данные из Teradata EDW через драйверы JDBC без потребности в каких-либо внешних шагах экспортирования данных (из СУБД) и их загрузки в Hadoop. TeradataInputFormat инспирирован подходом DBInputFormat [7], разработанным компанией Cloudera [6], но не основывается на нем. В отличие от подхода DBInputFormat, в котором каждый Mapper посылает в СУБД некоторый бизнес-запрос, представленный на SQL (и, таким образом, этот SQL-запрос выполняется столько раз, сколько имеется Mapper'ов Hadoop), коннектор TeradataInputFormat посылает в Teradata EDW бизнес-запрос только один раз, этот SQL-запрос выполняется только единожды, и каждый Mapper в параллель получает некотрую часть результатов прямо из узлов Teradata EDW.
• Мы обеспечиваем табличную UDF (User Defined Function – определяемая пользователями функция), которая при вызове из любого стандартного SQL-запроса выполняется в каждом параллельном компоненте Teradata EDW для параллельной выборки данных Hadoop прямо из узлов Hadoop. Любые реляционные таблицы можно соединить с данными Hadoop, выбираемыми этой табличной UDF, и любое средство бизнес-анализа, обеспечиваемое процессором SQL Teradata, можно применить как к реляционным данным, так и к данным Hadoop. Не требуются какие-либо внешние шаги для экспортирования данных Hadoop и их загрузки в Teradata EDW.

Оставшаяся часть статьи организована следующим образом. В разд. 2, 3 и 4 мы обсуждаем по очереди три вышеупомянутых подхода. В разд. 5 мы обсуждаем родственные работы. Разд. 6 содержит заключение.

2. Параллельная загрузка данных Hadoop в Teradata EDW

В этом разделе мы представляем подход DirectLoad, который мы разработали для эффективной параллельной загрузки данных Hadoop в Teradata EDW. Сначала мы кратко описываем утилиту/протокол FastLoad [2], широко используемую в производственных условиях для загрузки данных в таблицы Teradata EDW. Клиент FastLoad, прежде всего, подключается к процессу Gateway, выполняющемуся в одном из узлов системы Teradata EDW, которая представляет из себя кластер узлов. Клиент FastLoad образует столько сессий, сколько указывается пользователем Teradata EDW. Каждый узел в системе Teradata EDW конфигурируется таким образом, что в нем выполняется несколько виртуальных параллельных компонентов, называемых AMP (Access Module Processor – процессор модуля доступ) [2]. В Teradate AMP является единицей параллелизма; он отвечает за выполнение сканирования, соединений и других задач управления данными над данными, которыми он управляет. Каждая сессия управляется одним AMP, и число сессий, образуемых клиентом FastLoad, Teradata EDW не может превосходить число AMP. Программное обеспечение Teradata Gateway является интерфейсом между Teradata EDW и клиентами, подключенными к сети. Процессы Teradata Gateway обеспечивают коммуникации и управляют ими, а также сообщениями клиентов и шифрованием.

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

При загрузке в Teradata EDW разделенного файла DFS, сохраняемого в нескольких узлах Hadoop, возникают возможности оптимизации, которые отсутствуют при использовании СУБД, выполняемой в одном SMP-узле, или традиционного подхода FastLoad. Основная идея нашего подхода DirectLoad состоит в устранении двух пересылок данных, присутствующих в существующем подходе FastLoad. Первая пересылка выполняется от процесса Gateway к AMP-получателю, а вторая – от AMP-получателя к целевому AMP. В нашем подходе DirectLoad клиенту разрешается посылать данные в любой AMP-получатель, указываемый клиентом DirectLoad (в отличие от циклического подхода, реализованного в FastLoad). Поэтому мы можем устранить пересылку от Gateway к AMP-получателю за счет использования только AMP-получателей в том же узле, к которому подключен клиент DirectLoad.

Для описания того, как работает подход DirectLoad, мы используем следующий простейший случай. Сначала мы решаем, какую часть файла Hadoop должен получить каждый AMD, а затем образуем столько заданий DirectLoad, сколько AMD имеется в Teradata EDW. Каждое задание DirectLoad подключается к некоторому процессу Gateway, читает назначенную ему часть файла Hadoop с использованием API Hadoop, и пересылает данные подключенному процессу Gateway, который посылает данные Hadoop только одному уникальному AMP в том же узле Teradata. Так можно сделать, потому что каждому заданию DirectLoad известно, к какому процессу Gateway/узлу он подключен, и он может попросить Teradata EDW обнаружить список AMD, поддерживаемых в том же узле.

Поскольку нас более всего интересует быстрая пересылка данных из Hadoop в Teradata EDW, мы делаем каждый AMD-получатель целевым AMD, управляющим полученными им строками. Таким образом, вычислять значения хэш-функции на строках не требуется, и вторая пересылка в подходе DirectLoad устраняется. Однако при этом мы поступаемся тем, что над загружаемыми данными Hadoop не строится какой-либо индекс. Задания DirectLoad можно сконфигурировать таким образом, чтобы они выполнялись в системе Hadoop или же в системе Teradata EDW. Мы опускаем здесь обсуждение того случая, когда пользователю не угодно запускать столько заданий DirectLoad, сколько имеется AMP.

Наши предварительные эксперименты показывают, что DirectLoad может существенно превзойти FastLoad по производительности. В тестовой системе, которую мы использовали для экспериментов, имелось 8 узлов. В каждом узле имелось 4 процессора Pentium IV 3.6 GHz, 4 гигабайта основной памяти и два устройства с жесткими дисками, выделенных для использования в Teradata. Два других дисковых устройства предназначались для использования операционной системой и системой Hadoop (версия 0.20.1). В одной и той же тестовой системе функционировали и Teradata EDW, и Hadoop. В каждом узле запускались два AMP, чтобы можно было с пользой применять оба дисковых устройства, выделенных для целей Teradata.

Мы выполнили два эксперимента. В обоих экспериментах в одном задании FastLoad для загрузки данных Hadoop в Teradata EDW использовались 16 сессий. В данной системе максимальное число сессий, которое могло бы иметь задание FastLoad, равняется 16, посколько имеется всего 16 AMP. В подходе DirectLoad имелось по два задания DirectLoad на один узел, и в каждом задании DirectLoad использовалась одна сессия для посылки данных в локальный AMD. В обоих экспериментах в подходе DirectLoad одновременно имелось 16 активных сессий. В первом эксперименте мы генерировали DFS-файл с одним миллиардом строк. В каждой строке имелось два столбца. Во втором эксперименте мы генерировали DFS-файл со 150 миллионами строк. В каждой строке имелось 20 столбцов. Все столбцы были целого типа. В обоих экспериментах подход DirectLoad оказался примерно в 2,1 раза быстрее подхода FastLoad. Мы планируем выполнить большее число экспериментов при других конфигурациях системы.

Содержание Вперёд