Лепихов А.В. , Соколинский Л.Б.

Обработка запросов в СУБД для кластерных систем.

Лепихов А.В. , Соколинский Л.Б.
Южно-Уральский государственный университет
22 июня 2009 г.

Источник: http://omega.sp.susu.ac.ru/books/sok/papers/sources/LepikhovS%2010.pdf

Аннотация
Работа посвящена проблеме эффективной обработки запросов в кластерных вычислительных системах. Представлен оригинальный подход к размещению и репликации данных на узлах кластерной системы. На основе этого подхо- да разработан метод балансировки загрузки. Предложен метод эффективной параллельной обработки запросов для кластерных систем, основанный на опи- санном методе балансировки загрузки. Приведены результаты вычислительных экспериментов и выполнен анализ эффективности предложенных подходов.
1 Введение
На сегодняшний день существует целый ряд систем баз данных, обеспечивающих параллельную обработку запросов. Системы баз данных, предназначенные для обра- ботки OLAP-запросов используются для управления петабайтными массивами дан- ных. Например, СУБД Greenplum основанная на технологии MapReduce [1] выпол- няет глубинный анализ 6.5 Пбайт данных на 96-узловом кластере в компании eBay. СУБД Hadoop обрабатывает 2.5 Пбайт данных на кластере, состоящем из 610 узлов для популярного web-сервиса facebook. В области параллельной обработки OLTP- запросов существует ряд коммерческих параллельных СУБД среди которых наибо- лее известными являются Teradata, Oracle Exadata и DB2 Parallel Edition.
В настоящее время исследования в данной области ведутся в направлении самона- стройки СУБД [2], балансировки загрузки и связанной с ней проблемы размещения данных [3], оптимизации параллельных запросов [4] и эффективного использования современных многоядерных процессоров [5, 6].
Одной из важнейших задач в параллельных СУБД является балансировка за- грузки. В классической работе [7] было показано, что перекосы, возникающие при обработке запросов в параллельных системах баз данных без совместного использо- вания ресурсов, могут приводить к практически полной деградации производитель- ности системы.
∗Работа выполнена при финансовой поддержке Российского фонда фундаментальных исследо- ваний (проект 09-07-00241-а).
Рис. 1: Фрагментный параллелизм.
В работе [8] предложено решение проблемы балансировки загрузки для систем без совместного использования ресурсов, основанное на репликации. Данное реше- ние позволяет уменьшить накладные расходы на передачу данных по сети в процессе балансировки загрузки. Однако этот подход применим в весьма узком контексте про- странственных баз данных в специфическом сегменте диапазонных запросов. В рабо- те [3] задача балансировки загрузки решается путем частичного перераспределения данных перед началом выполнения запроса. Данный подход уменьшает суммарное количество пересылок данных между вычислительными узлами в ходе обработки запроса, однако накладывает серьезные требования на скорость межпроцессорных коммуникаций.
В настоящей работе предложен метод параллельной обработки запросов, осно- ванный на оригинальном подходе к размещению базы данных, названном частичным зеркалированием. Данный метод решает задачи эффективной обработки запросов и балансировки загрузки в кластерных системах.
Статья организована следующим образом. В разделе 2 описывается метод парал- лельной обработки запросов в СУБД для кластерных систем. В разделе 3 предлагает- ся стратегия размещения данных на кластерных системах и алгоритм балансировки загрузки. В разделе 4 приведены результаты вычислительных экспериментов, кото- рые позволяют оценить практическую значимость предлагаемых в работе методов и алгоритмов. В заключении суммируются основные результаты и выводы, получен- ные в данной статье, и намечаются направления дальнейших исследований.
2 Организация параллельной обработки запросов
Основой параллельной обработки запросов в реляционных системах баз данных яв- ляется фрагментный параллелизм (см. рис. 1). Данная форма параллелизма предпо- лагает фрагментацию отношения, являющегося аргументом реляционной операции, по дискам многопроцессорной системы. Способ фрагментации определяется функ- цией фрагментации φ, которая для каждого кортежа отношения вычисляет номер процессорного узла, на котором должен быть размещен этот кортеж. Запрос па- раллельно выполняется на всех процессорных узлах в виде набора параллельных агентов [9], каждый из которых обрабатывает отдельный фрагмент отношения на выделенном ему процессорном узле. Полученные агентами результаты сливаются в результирующее отношение. Несмотря на то, что каждый параллельный агент в процессе выполнения запроса независимо обрабатывает свой фрагмент отношения, для получения корректного результата необходимо выполнять пересылки кортежей.
Рис. 2: Схема обработки запроса в параллельной СУБД для кластерных систем. Q – последовательный физический план, Ai – параллельный агент, C Ni – вычислитель- ный узел.
Для организации таких пересылок в соответствующие места дерева плана запроса вставляется оператор exchange [10].
Оператор exchange идентифицируется в дереве плана своим номером и функцией распределения ψ, которая для каждого входного кортежа вычисляет номер вычис- лительного узла, где должен быть обработан данный кортеж. Оператор exchange выполняет пересылки кортежей между параллельными агентами, используя комму- никационные каналы, каждый из которых задается парой (номер узла, номер порта). При этом в качестве номера узла фигурирует номер параллельного агента, а в каче- стве порта – номер оператора exchange. Опишем общую схему организации параллельной обработки запросов в парал- лельной СУБД для кластерных систем. Мы предполагаем, что вычислительная си- стема представляет собой кластер, состоящий из N вычислительных узлов (см. рис. 2). Будем считать, что каждое отношение базы данных, задействованное в обработке за- проса, фрагментировано по всем узлам вычислительной системы. В соответствии с данной схемой, обработка SQL-запроса состоит из трех этапов. На первом этапе SQL-запрос передается пользователем на выделенную host-машину, где транслируется в некоторый последовательный физический план. На втором этапе последовательный физический план преобразуется в параллельный план, представ- ляющий собой совокупность параллельных агентов. Это достигается путем вставки оператора обмена exchange в соответствующие места дерева запроса. На третьем этапе параллельные агенты пересылаются с host-машины на соот- ветствующие вычислительные узлы, где интерпретируются исполнителем запросов. Результаты выполнения агентов объединяются корневым оператором exchange на нулевом узле, откуда передаются на host-машину. Роль host-машины может играть любой узел вычислительного кластера.
Поясним цикл обработки запроса в параллельной системе баз данных на следу- ющем примере. Пусть необходимо вычислить Q = R ./ S – естественное соединение двух отношений R и S по некоторому общему атрибуту Y . Пусть отношение R фраг- ментировано по атрибуту соединения на двух вычислительных узлах C N0 и C
виде двух фрагментов R0 и R1, т.е. R = R0 ∪ R1, R0 ∩ R1 = ∅, πY (R0) ∩ πY (R1) = ∅,
где π – операция проекции. Обозначим φ : R → {0, 1} – функция фрагментации для отношения R. Имеем ∀u, v ∈ R : u.Y = v.Y ⇒ φ(u) = φ(v).
Здесь u.Y и v.Y обозначают значение атрибута Y в кортежах u и v соответственно. Тогда существует функция φY : πY (R) → {0, 1} такая, что ∀r ∈ R : φ(r) = φY (r.Y ).
Пусть отношение S фрагментировано по некоторому другому атрибуту Z на тех же двух вычислительных узлах C N0 и C N1 в виде двух фрагментов S0 и S1, т.е. S = S0 ∪ S1, S0 ∩ S1 = ∅, πZ (S0) ∩ πZ (S1) = ∅, Z = Y.
Тогда последовательный физический план запроса Q и соответствующий ему па- раллельный план будут иметь вид, изображенный на рис. 3. Параллельный план в
данном случае включает в себя двух агентов A0 и A1, которые будут выполняться на вычислительных узлах C N0 и C N1 соответственно. Для того, чтобы естественное соединение выполнялось в параллельном плане корректно, нам необходимо вставить оператор exchange e1 между оператором join и оператором scan для отношения S. При этом функция распределения для e1 будет иметь вид
ψ1(s) = φY (s.Y ).
Для сбора кортежей результирующего отношения на узле агента A0 после опера- тора join добавляем еще один оператор exchange e2, функция распределения кото- рого имеет вид ψ2(x) = 0.
3 Размещение данных и балансировка загрузки
3.1 Фрагментация и сегментация данных
Распределение базы данных в кластерной вычислительной системе задается следу- ющим образом [11]. Каждое отношение разбивается на непересекающиеся горизон- тальные фрагменты, которые размещаются на различных вычислительных узлах. При этом предполагается, что кортежи фрагмента некоторым образом упорядоче- ны, что этот порядок фиксирован для каждого запроса и определяет последователь- ность считывания кортежей в операции сканирования фрагмента. Будем называть этот порядок естественным. На практике естественный порядок может определять- ся физическим порядком следования кортежей или индексом.
Каждый фрагмент на логическом уровне разбивается на последовательность сег- ментов фиксированной длины. Длина сегмента измеряется в кортежах и является атрибутом фрагмента. Разбиение на сегменты выполняется в соответствии с есте- ственным порядком и всегда начинается с первого кортежа. В соответствии с этим последний сегмент фрагмента может оказаться неполным. Количество сегментов фрагмента F обозначается как S(F ) и может быть вычис- лено по формуле

S(F ) =
» T (F ) ј . L(F )
Здесь T (F ) обозначает количество кортежей во фрагменте F , L(F ) – длину сег- мента для фрагмента F .
3.2 Репликация данных
Пусть фрагмент F0 располагается на диске d0 ∈ D кластерной системы. Полагаем, что на каждом диске di ∈ D(i > 0) располагается частичная реплика Fi, включаю- щая в себя некоторое подмножество (возможно пустое) кортежей фрагмента F0. Наименьшей единицей репликации данных является сегмент. Длина сегмента ре- плики всегда совпадает с длиной сегмента реплицируемого фрагмента:
L(Fi) = L(F0), ∀di ∈ D.
Рис. 4: Параллельный агент с двумя входными потоками.
Размер реплики Fi задается коэффициентом репликации ρi ∈ R, 0 ≤ ρi ≤ 1,
являющимся атрибутом реплики Fi, и вычисляется по следующей формуле T (Fi) = T (F0) − d(1 − ρi) • S(F0)e • L(F0).
Естественный порядок кортежей реплики Fi определяется естественным порядком кортежей фрагмента F0 . При этом номер N первого кортежа реплики Fi вычисляется по формуле: N (Fi) = T (F0) − T (Fi) + 1.
Для пустой реплики Fi будем иметь N (Fi) = T (F0)+1, что соответствует признаку ѕконец файлаї.
Описанный механизм репликации данных позволяет использовать в кластерных системах простой и эффективный метод балансировки загрузки, описываемый в раз- деле 3.3.
3.3 Метод балансировки загрузки
3.3.1 Схема работы параллельного агента
Пусть задан некоторый запрос Q, имеющий n входных отношений. Пусть Q – парал- лельный план запроса Q. Каждый агент Q ∈ Q имеет n входных потоков s1, . . . , sn. Каждый поток si(i = 1, . . . , n) задается четырьмя параметрами:
1. fi – указатель на фрагмент отношения;
2. qi – количество сегментов в отрезке, подлежащем обработке;
3. bi – номер первого сегмента в обрабатываемом отрезке;
4. ai – индикатор балансировки: 1 – балансировка допускается, 0 – балансировка не допускается.
На рис. 4 изображен пример параллельного агента с двумя входными потоками. Параллельный агент Q может находиться в одном из двух состояний: активном и пассивном. В активном состоянии Q последовательно считывает и обрабатывает кортежи из всех входных потоков. При этом в ходе обработки динамически изме- няются значения параметров qi и bi для всех i = 1, . . . , n. В пассивном состоянии агент Q не выполняет никаких действий. На начальном этапе выполнения запроса выполняется инициализация агента, в результате которой происходит определение параметров всех входных потоков. Затем агент переводится в активное состояние и начинает обработку фрагментов, ассоциированных с его входными потоками. В каждом фрагменте обрабатываются только те сегменты, которые входят в отрезок, определяемый параметра�