Новые технологии распределенного хранения и обработки больших массивов данных о. В. Сухорослов
 Скачать 296.71 Kb.
|
|
1 НОВЫЕ ТЕХНОЛОГИИ РАСПРЕДЕЛЕННОГО ХРАНЕНИЯ И ОБРАБОТКИ БОЛЬШИХ МАССИВОВ ДАННЫХ О.В. Сухорослов Институт системного анализа РАН 117312, г. Москва, пр-т 60-летия Октября, д. 9 Аннотация. В статье рассматриваются новые технологии, позволяющие организовать распределенное хранение и параллельную обработку больших объемов данных в крупномасштабных кластерных системах. Речь идет о петабайтах данных, для хранения и обработки которых необходимы значительные вычислительные ресурсы. В качестве таких ресурсов рассматриваются кластерные системы, состоящие из тысяч серверов. В подобных распределенных системах остро стоят вопросы обеспечения отказоустойчивости и бесперебойного функционирования сервисов хранения и обработки данных. Другой важной проблемой является создание высокоуровневой модели программирования процессов обработки данных на подобных системах, скрывающей от пользователя детали распределения данных и планирования вычислений в ненадежной распределенной среде. В статье приводится описание оригинальных технологий, нацеленных на решение указанных проблем и уже применяемых в крупнейших информационных системах. Поскольку большинство подобных технологий являются закрытыми коммерческими разработками, особое внимание уделено описанию создаваемых в настоящее время открытых (open source) аналогов данных технологий. Annotation. The article describes emerging technologies for distributed storage and processing of large data sets on cluster systems. These technologies enable parallel processing of petabytes of data on thousands of machines. Such large-scale distributed computing systems pose some challenges concerning enabling automatic fault-tolerance and high availability of services. Another important issue is a development of high-level programming 2 models and data processing tools which hide from a user all low-level details of data distribution, scheduling and process synchronization in a distributed environment. The article presents several emerging technologies aimed on solving the above-mentioned problems. Some of these, mostly proprietary, technologies are already used in largest commercial information systems, while others provide open source analogues of closed proprietary solutions. 3 Введение В современном мире все большую роль играют технологии, обеспечивающие эффективное хранение и обработку данных. Связано это с наблюдаемым с конца прошлого века лавинообразным ростом информации. Современные задачи и приложения, связанные с анализом данных, предъявляют особые требования к вычислительным ресурсам, значительно превышающие возможности отдельных компьютеров. Описываемые в статье технологии берут свое начало внутри компании Google, крупнейшей поисковой системы Web. В настоящее время Web насчитывает десятки миллиардов страниц. Роботы поисковой системы круглосуточно загружают петабайты данных с содержанием новых и измененных Web-страниц. Загруженные данные подвергаются различным процедурам обработки, связанным с построением индекса Web, вычислением индексов цитирования отдельных страниц и т.д. [1]. Полученные в результате обработки данные, измеряемые также петабайтами, размещаются в долговременном хранилище и используются для генерации результатов поисковых запросов пользователей. В настоящее время появляется все больше Web-приложений, накапливающих и фильтрующих большие объемы информации, таких как социальные сети и различные сервисы агрегации. Вычислительная инфраструктура Google насчитывает сотни тысяч серверов, размещенных в нескольких центрах обработки данных по всему миру. По опубликованной статистике [2] в сентябре 2007 г. в инфраструктуре Google было запущено в общей сложности более 2 млн. заданий со средним количеством задействованных машин около 400. Суммарный объем входных данных заданий составил около 400 петабайт. Другой областью, где распространены приложения, связанные с хранением и обработкой больших объемов данных, являются научные вычисления. Чаще всего это выражается в необходимости проведения трудоемкого анализа собранных массивов экспериментальных данных для получения новых научных результатов. Ярким примером является крупномасштабный международный проект в области физики высоких энергий по созданию Большого адронного коллайдера (БАК). Предполагается, что во время проведения экспериментов БАК будет генерировать 4 порядка 15 петабайтов первичных экспериментальных данных в год. Для хранения и обработки этих данных планируется использовать специально созданную Grid- инфраструктуру WLCG/EGEE, насчитывающую более 200 ресурсных центров в десятках стран с общей численностью в тысячи машин. Подобного масштаба задачи встречаются и в других областях исследований, таких как биоинформатика, астрофизика, науки о Земле и т.д. Для достижения приемлемого времени работы подобным приложениям необходимы ресурсы уровня суперкомпьютеров или кластеров с сотнями и тысячами узлов. Несмотря на значительный рост емкости носителей данных, временные характеристики доступа к данным изменились слабо, что приводит к необходимости кластеризации носителей для обеспечения приемлемой пропускной способности. Для обеспечения надежности хранения данных также необходима избыточная репликация данных. Постоянный рост объемов обрабатываемых данных требует соответствующего наращивания вычислительных ресурсов, в связи с чем используемая вычислительная среда должна обладать высокой масштабируемостью. В настоящее время в области массированной обработки данных наблюдается переход от специализированных суперкомпьютерных архитектур к более экономичным и масштабируемым, но менее надежным кластерным системам из недорогих серверов массового производства. Реализация процедуры обработки данных в подобной распределенной среде сопряжена с решением таких непростых задач, как разбиение и распределение данных между процессорами, планирование и балансировка нагрузки, обработка отказов отдельных узлов, сбор и агрегация промежуточных результатов. Необходимость непосредственной реализации данных механизмов при программировании процедуры обработки данных является серьезным препятствием на пути широкого внедрения подобных систем. Поэтому необходимо, чтобы соответствующие технологии уже содержали в себе реализации данных механизмов и предоставляли пользователю высокоуровневые модели программирования, скрывающие от него детали реализации вычислений в ненадежной распределенной среде. 5 1. Распределенные системы хранения данных В главе рассматриваются новые системы, ориентированные на хранение больших массивов данных в распределенных кластерных системах. Условно их можно разделить на два класса: распределенные файловые системы (Google File System, Hadoop Distributed System) и распределенные хранилища структурированных данных (Google BigTable, HBase). Рассматриваемые системы имеют принципиальные отличия от традиционных файловых систем и реляционных баз данных. 1.1. Google File System Распределенная файловая система Google File System (GFS) является закрытой разработкой компании Google, используемой для хранения больших массивов данных. Внутри Google функционирует более 200 GFS-кластеров, крупнейшие из которых насчитывают более 5 тысяч машин, хранящих около 5 петабайт данных и обслуживающих порядка 10 тысяч клиентов [3]. Описание GFS было опубликовано в работе [4]. Как и любая распределенная файловая система, GFS ориентирована на обеспечение высокой производительности, масштабируемости, надежности и доступности. Отличия архитектуры GFS от других распределенных файловых систем, таких как AFS, xFS, Lustre, обусловлены спецификой приложений и вычислительной инфраструктуры Google. Данная инфраструктура построена из большого количества недорогих серверов массового производства, что приводит к постоянным отказам оборудования. Поэтому внутренние технологии Google в обязательном порядке предусматривают механизмы обнаружения и автоматического восстановления после отказов отдельных машин. Специфика приложений Google такова, что хранимые файлы обычно имеют большой размер (многие гигабайты) по сравнению с обычными файловыми системами. Как правило, содержимое файлов изменяется только за счет записи новых данных в конец файла. Запись может вестись одновременно несколькими клиентами, поэтому требуются гарантии атомарности отдельных операций. После окончания записи, файлы в основном только считываются, причем – последовательно. 6 Операции чтения больших порций данных могут происходить в потоковом режиме. Стабильно высокая пропускная способность более важна для приложений, чем низкое время отклика при выполнении отдельных операций. Файлы в GFS организованы в иерархическую структуру директорий и идентифицируются при помощи полных путей. GFS предоставляет интерфейс с типичными для файловой системы операциями create, delete, open, close, read и write. Дополнительно поддерживаются операции snapshot и record append. Первая операция создает копию файла или директории. Вторая операция реализует атомарную запись в конец файла. В отличие от многих файловых систем, в целях упрощения реализации и повышения эффективности GFS не реализует стандартный POSIX-интерфейс. GFS – распределенная система, организованная по схеме “главный- подчиненный” (master-slave). В системе есть один главный сервер (master) и несколько chunk- серверов. Файлы разбиваются на блоки (chunk) фиксированного размера (обычно 64 Мб), размещаемые на chunk-серверах. Для обеспечения надежности и распределения нагрузки каждый блок реплицируется на нескольких (по умолчанию, трех) серверах, размещенных в различных серверных стойках. Масштабируемость системы достигается за счет возможности “горячего” подключения новых chunk-серверов, а также описываемых ниже стратегий, позволяющих разгрузить главный сервер и, тем самым, избежать возникновения в системе узкого места. Главный сервер хранит в памяти все метаданные файловой системы, включая пространства имен, права доступа, отображение файлов в блоки и текущие местоположения всех реплик блоков. Главный сервер также контролирует общесистемные процессы, такие как размещение и репликация блоков, назначение главной реплики, удаление неиспользуемых блоков и миграция блоков между серверами. Chunk-сервера периодически отправляют главному серверу сообщения с информацией о своем состоянии. Главный сервер использует ответы на эти сообщения для передачи инструкций chunk-серверам. GFS- клиент реализует интерфейс (API) файловой системы и взаимодействует с главным и chunk-серверами от имени приложения. Важно отметить, что клиент обращается к главному серверу только за метаданными, а все операции по чтению и 7 записи данных осуществляются напрямую с chunk-серверами. Это позволяет уменьшить нагрузку на главный сервер. Рассмотрим схему взаимодействия клиента с GFS на примере чтения файла. Сначала клиент вычисляет по указанному байтовому отступу в файле номер соответствующего блока, после чего отправляет главному серверу запрос с именем файла и номером блока. В ответ главный сервер передает клиенту уникальный идентификатор данного блока и адреса его реплик. Клиент кэширует эту информацию локально, после чего отправляет запрос одному (например, ближайшему) из chunk- серверов с репликой блока. В запросе клиент указывает идентификатор блока и байтовый отступ внутри блока. Дальнейшие операции чтения данных из этого блока осуществляются клиентом автономно, без участия главного сервера, до тех пор пока информация в кэше не будет удалена по истечении времени жизни или при повторном открытии файла. В целях дополнительно оптимизации нагрузки, клиент обычно запрашивает у главного сервера информацию сразу о нескольких блоках. GFS реализует ослабленную модель целостности данных, достаточную для работы целевых приложений и, в то же время, относительно простую и эффективную в реализации. Операции изменения метаданных, такие как создание файлов, являются атомарными и выполняются главным сервером. Операции изменения данных разделяются на два типа: запись с определенным отступом (write) и присоединение (record append). В первом случае не гарантируется атомарность выполнения операции, то есть записанные одним клиентом данные могут быть “перемешаны” с данными другого клиента с несоблюдением указанных отступов. Но при этом гарантируется, что все клиенты будут видеть одни и те же данные, независимо от используемой реплики. Во втором случае гарантируется, что данные будут записаны атомарно (непрерывным фрагментом) как минимум один раз. При этом внутри данного региона файла могут появляться неполные фрагменты и дубликаты записываемых данных, связанные с возникновением ошибок при записи данных на уровне реплик. В этом случае GFS не гарантирует, что все реплики будут абсолютно идентичны, а лишь - что присоединяемые данные будут записаны во все реплики атомарно. Описанная модель целостности накладывает обязательства на клиентов GFS по проверке считываемых данных на наличие неполных фрагментов и дубликатов. Подобную проверку можно 8 реализовать с помощью вставки в записываемые данные контрольных сумм и идентификаторов записей. В целях снижения нагрузки на главный сервер, операции записи данных в блок координируются одним из chunk-серверов, хранящих реплику блока. Данная реплика называется главной (primary). Центральный сервер управляет назначением центральных реплик путем выдачи реплике временной аренды (lease) с возможностью ее продления. Главная реплика управляет порядком внесения изменений при одновременной записи данных в ее блок, координируя действия остальных, подчиненных (secondary) реплик блока. При записи данных клиент получает от главного сервера адрес главной реплики блока, после чего взаимодействует напрямую с chunk- серверами. Клиент отправляет данные на все реплики блока и, после получения подтверждений от реплик, обращается в главной реплике с запросом на запись данных. Главная реплика присваивает каждому поступающему запросу номер, применяет изменения к локальным данным в соответствии с номером запроса и передает запрос остальным репликам с указанием его номера. После получения уведомлений о выполнении запроса от подчиненных реплик, главная реплика возвращает ответ клиенту. В случае возникновения ошибок при записи данных, клиент пытается повторить описанную процедуру повторно. Для эффективного использования сетевых ресурсов, записываемые данные передаются репликам путем их пересылки по цепочке от одного chunk-сервера к другому. При получении первой порции данных сервер тут же начинает их передачу в конвейерном режиме ближайшему (в терминах топологии систем) серверу. Это позволяет полностью задействовать пропускную способность каждой машины, избежать возникновения узких мест и минимизировать задержку при передаче данных. GFS поддерживает операцию shapshot, позволяющую быстро создавать копии файлов или целых директорий. Эта операция используется приложениями для ветвления наборов данных или сохранения копии данных перед их изменением, с возможностью последующего отката к старой версии. Для реализации snapshot используется стандартная стратегия отложенного копирования (copy-on-write) – копии данных создаются во время первой последующей записи в копируемые данные. При 9 выполнении операции главный сервер отзывает все аренды у главных реплик блоков, входящих в копируемые данные, что позволяет ему контролировать операции записи. Главный сервер GFS управляет размещением реплик в системе. При создании нового блока он назначает chunk-сервера для хранения реплик блока исходя из такой информации о серверах, как объем используемого дискового пространства, число недавно размещенных на сервере реплик, положение в сетевой топологии. Главный сервер постоянно контролирует состояние chunk-серверов и хранимых на них реплик с помощью сообщений, получаемых от серверов. В случае выхода из строя сервера или повреждения реплики, главный сервер автоматически выполняет размещение новых реплик до тех пор, пока не будет восстановлен заданный уровень репликации данных. Аналогичная процедура проводится в случае, если пользователь увеличивает уровень репликации хранимых данных. Кроме того, главный сервер периодически производит балансировку системы, перераспределяя реплики между chunk-серверами. Наконец, главный сервер производит периодическую “сборку мусора”, заключающуюся в физическом удалении формально удаленных или поврежденных данных. Как уже упоминалась, вычислительная инфраструктура Google подвержена постоянным и неизбежным отказам оборудования. GFS учитывает это обстоятельство путем реализации механизма автоматического восстановления после отказов. Главный и chunk-сервера реализованы таким образом, чтобы максимально быстро восстанавливать свое состояние после аварийного перезапуска. Главный сервер сохраняет все операции и изменения своего состояния в log- файле, который автоматически реплицируется на нескольких машинах. При превышении максимального размера log-файла создается резервная копия состояния главного сервера, а log-файл очищается. В случае аварийной остановки главного сервера, он автоматически перезапускается и восстанавливает свое состояние по последней резервной копии и log-файлу. Также некоторое время уходит на получение главным сервером информации о местоположении реплик от chunk-серверов, поскольку данная информация не сохраняется главным сервером. В случае отказа самой машины, главный сервер автоматически запускается на другой машине кластера и восстанавливает состояние с помощью реплицированных файлов. Клиенты автоматически устанавливают соединение с новым сервером по истечении таймаута, 10 поскольку главный сервер адресуется с помощью DNS-имени. В системе также присутствует несколько резервных главных серверов (shadow master), которые предоставляют доступ к системе в режиме чтения, даже если главный сервер недоступен. В случае аварийной остановки главного сервера, он автоматически перезапускается и восстанавливает свое состояние по хранимым на диске данным. В случае отказа самой машины с chunk-сервером, главный сервер автоматически удаляет ссылки на хранимые сервером реплики и начинает процесс их восстановления с помощью реплик на других серверах. Напомним, что каждый блок по умолчанию хранится на трех серверах. На chunk-серверах также могут случаться сбои жестких дисков, приводящие к частичному повреждению хранимых реплик. Поэтому каждый chunk- сервер выполняет проверку целостности хранимых им данных с помощью контрольных сумм. В случае если контрольная сумма не совпадает, то chunk-сервер сообщает об этом главному серверу, который создает новую реплику блока. В заключении отметим главные особенности распределенной файловой системы GFS: Высокая отказоустойчивость системы, автоматическое восстановление после отказов и поддержка кластеров из массовых серверов; Ориентация на относительно небольшое число крупных файлов, которые записываются однократно; Оптимизация под операции записи в конец файла, выполняемые одновременно многими клиентами; Нестандартный интерфейс файловой системы и ослабленная модель целостности данных; Эффективное использование сетевых ресурсов и оптимизация под высокую агрегированную пропускную способность. |