Hadoop: введение в системы больших данных

VPS | Amber | Комментировать запись

Apache Hadoop – один из важнейших открытых инструментов для хранения и обработки большого количества цифровых данных, накопленных с ростом World Wide Web. Он развился из открытого проекта под названием Nutch, который предназначался для поиска в Интернете. Создатели Nutch были в большой степени подвержены влиянию Google. В конечном итоге функции хранения и обработки были  выделены в проект Hadoop, а Nutch разрабатывается как инструмент поиска.

Данная статья расскажет, что такое системы больших данных.

Системы данных

Данные существуют повсюду: на клочках бумаги, в книгах, на фотографиях, в мультимедийных файлах, логах сервера и на веб-сайтах. Когда эти данные собираются целенаправленно, они входят в систему данных.

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

Этот проект хорошо иллюстрирует систему данных:

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

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

Компании поисковых систем столкнулись с этой конкретной проблемой, поскольку количество веб-контента стало стремительно расти в эпоху Dot-com. В 2003 году Google опубликовал статью The Google File System , где описывается, как их закрытое программное обеспечение обрабатывало огромное количество данных поисковой системы. В 2004 году Google публикует MapReduce: Simplified Data Processing on Large Clusters, где подробно описывается, как кластеры упрощают обработку таких больших объемов данных. Эти две статьи сильно повлияли на архитектуру Hadoop.

Чем отличаются большие данные?

Статьи Google и реализация этих идей в Hadoop основаны на четырех изменениях в восприятии данных, которые необходимы для учета объема данных:

1. Системы больших данных должны поддерживать распределение данных. Распределенное хранение набора данных на разных машинах стало неизбежным.
2. Когда кластеры стали основой хранилища, программное обеспечение должно было научиться учитывать аппаратный сбой, поскольку это неизбежно, особенно если речь идет о сотнях или тысячах компьютеров в кластере.
3. Поскольку на машинах будут случаться сбои, им нужен новый способ общения друг с другом. В повседневных вычислениях данных машины обычно определяются IP-адресом или именем хоста. Это явное сообщение машин пришлось заменить неявным соединением: при этом одна машина сообщает какой-то другой машине, что она должна обрабатывать некоторые конкретные данные. В противном случае программисты столкнулись бы с проблемой аутентификации – такой же большой, как и сама проблема обработки данных.
4. Компьютеру нужно будет перейти к данным и обработать их на распределенных машинах, а не перемещать огромное количество данных по сети.

Выпущенная в 2007 году версия 1.0 основанного на Java фреймвока Hadoop стала первым открытым проектом, который учитывал все эти изменения. Его первая версия состоит из двух уровней:

• HDFS: распределенная файловая система Hadoop, которая отвечает за хранение данных на нескольких компьютерах.
• MapReduce: программная среда для параллельной обработки данных на каждой машине, а также для планирования задач, их мониторинга и перезапуска .

HDFS 1.0

Распределенная файловая система Hadoop, HDFS, представляет собой уровень хранения, который Hadoop использует для распространения и надлежащего хранения данных для обеспечения высокой доступности.

Как работает HDFS 1.0?

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

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

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

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

Ограничения HDFS 1.0

HDFS 1.0 сделал Hadoop лидером среди открытых инструментов для хранения больших данных. Отчасти этот успех был вызван решениями в архитектуре, которые упростили распределенное хранение. Но при этом ограничения оставались. К основным ограничениям версии 1.0 относятся:

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

Несмотря на эти ограничения, HDFS сделал большой вклад в работу с большими данными.

MapReduce 1.0

Второй уровень Hadoop – MapReduce – отвечает за пакетную обработку данных, хранящихся на HDFS. Внедрение в Hadoop модели Google MapReduce позволяет разработчикам использовать ресурсы HDFS без параллельных и распределенных систем.

Как работает MapReduce 1.0

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

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

Давайте рассмотрим, как это работает на таком примере:

She sells seashells by six seashores.
She sure sells seashells well.
MAPPING         SHUFFLING           REDUCING
{she, 1}        {she, 1, 1}         {she, 2}
{sells, 1}      {sells, 1, 1}       {sells, 2}
{seashells, 1}  {seashells, 1, 1}   {seashells, 2}
{by, 1}         {by, 1}             {by, 1}
{six, 1}        {six, 1}            {six, 1}
{seashores, 1}  {seashores, 1, 1}   {seashores, 2}
{she, 1}        {sure, 1}           {sure, 1}
{sure, 1}       {well, 1}           {well, 1}
{sells}
{seashells, 1}
{well, 1}

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

Компоненты более высокого уровня могут подключаться к MapReduce для предоставления дополнительных функций. Например, Apache Pig предоставляет разработчикам язык для написания программ анализа данных, абстрагируя идиомы Java MapReduce на более высокий уровень (аналогично тому, что делает SQL для реляционных баз данных). Apache Hive поддерживает анализ данных и отчетность с помощью SQL-подобного интерфейса для HDFS. Он абстрагирует запросы MapReduce Java API для обеспечения функциональности запросов высокого уровня. Для Hadoop 1.x доступно множество дополнительных компонентов, но экосистема MapReduce также имеет некоторые ограничения.

Ограничения MapReduce 1

• Зависимость между MapReduce и HDFS. В реализации 1.x обязанности уровня MapReduce выходят за рамки обработки данных (включая управление ресурсами кластера) и тесно связаны с HDFS. Это означает, что разработчикам аддонов для 1.x приходится писать многопроходные программы MapReduce, независимо от того, подходит ли это для задачи или нет, потому что MapReduce является единственным способом доступа к файловой системе.
• Статические слоты для анализа данных. Сопоставление и сокращение происходит на DataNodes, но на каждом DataNode доступно только ограниченное статическое количество одноцелевых слотов. Слоты мапперов могут только сопоставлять, а слоты сокращения – только сокращать. Число слотов устанавливается в конфигурации без возможности динамической настройки, и поэтому могут возникнуть ситуации, когда рабочая нагрузка кластера не соответствует конфигурации. Такое распределение слотов также затрудняет планирование других приложений.
• JobTracker – единая точка отказа. Приложения Hadoop отправляют задачи MapReduce в JobTracker, который, в свою очередь, распределяет эти задачи на определенные ноды кластера либо по доступным слотам, либо по географической близости. TaskTracker уведомляет JobTracker, если задача выполнена с ошибкой. JobTracker может повторно отправить задачу, пометить запись, которая будет исключена из будущей обработки, или поместить в черный список ненадежный TaskTracker. Но в случае сбоя самого JobTracker все задачи MapReduce будут остановлены.

Улучшения в Hadoop 2.x

Ветка Hadoop 2.х, выпущенная в декабре 2011 года, представила четыре основных усовершенствования и исправила ключевые ограничения версии 1. Hadoop 2.0 устраняет ограничение производительности и единую точку отказа NameNode. Кроме того, он отделяет MapReduce от HDFS с введением YARN (Yet Another Resource Negotiator), открыв экосистему дополнительных продуктов и разрешив моделям обработки взаимодействовать с HDFS и обходить слой MapReduce.

1: Федерация HDFS

Федерация HDFS вводит четкое разделение пространства имен и хранилища, что делает возможным наличие нескольких пространств имен в кластере. Благодаря этому появляются такие улучшения:

• Масштабирование пространства имен. Возможность добавить больше имен в кластер обеспечивает горизонтальное масштабирование. Большим кластерам или кластерам со множеством небольших файлов может быть полезно добавить дополнительные NameNodes.
• Производительность. Наличие нескольких NameNode снимает ограничение на операции файловой системы.
• Изоляция между пространствами имен. В многопользовательских средах с одним NameNode один пользователь мог влиять на каждого другого пользователя в системе. В федерации стало возможным изолировать жителей системы.

Как работает федерация HDFS

Федерация управляет пространством имен файловой системы. NameNodes работают независимо и не координируются друг с другом. DataNodes регистрируются в кластере с каждым NameNode, отправляют пульс и отчеты блоков и обрабатывают входящие команды из NameNode.

Блоки распространяются по всему хранилищу с той же случайной репликацией, что и в Hadoop 1.x. Все блоки, принадлежащие одному пространству имен, называются пулом блоков. Такие пулы управляются независимо, позволяя пространству имен генерировать идентификаторы блоков для новых блоков без согласования с другими пространствами имен. Комбинация пространства имен и пула блоков называется томом пространства имен; том формирует автономный блок, так что когда один из NameNode удаляется, его пул блоков удаляется вместе с ним.

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

2: Высокая доступность NameNode

Если в предыдущих версиях NameNode прекращал работу, весь кластер был недоступен, пока NameNode не перезапустится или не появится на новом компьютере. Модернизация программного или аппаратного обеспечения NameNode также создавала окна простоя. Чтобы предотвратить это, Hadoop 2.0 реализовал конфигурацию active/passive, чтобы обеспечить быстрый переход на другой ресурс.

Как работает высокая доступность NameNode

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

3: YARN

Hadoop 2.0 отделяет MapReduce от HDFS. Управление рабочими нагрузками, многоуровневым обслуживанием, безопасностью и функциями высокой доступности было выделено в YARN (Yet Another Resource Negotiator). YARN – это, по сути, крупномасштабная распределенная операционная система для приложений больших данных, которая позволяет использовать Hadoop как для MapReduce, так и для других приложений, которые не могут дождаться завершения пакетной обработки. YARN устранил необходимость работы через инфраструктуру MapReduce с высокой задержкой ввода-вывода, что позволяет использовать новые модели обработки HDFS.

У пользователей Hadoop 2.x есть доступ к таким моделям обработки.

• Пакетная обработка. Системы пакетной обработки не являются интерактивными и имеют доступ ко всем данным до начала обработки. Пакетная обработка обычно выполняется с высокой задержкой, причем скорость выполнения заданий пакетной обработки больших данных обычно измеряется в минутах или более. Эта модель подходит для индексирования данных, поиска в Интернете и обработки данных. Программы, которые могут сделать это для Hadoop: MapReduce, Tez, Spark, Hive и Flink.
• Интерактивная обработка. Системы интерактивной обработки необходимы, когда вопросы неизвестны заранее. Вместо этого пользователь интерпретирует ответ на запрос, а затем формулирует новый вопрос. Для поддержки такого рода обработки ответ нужно вернуть гораздо быстрее, чем при обычной работе MapReduce. Эта модель подходит для просмотра данных. Программы, которые могут сделать это для Hadoop: Impala, Drill, HAWQ, Presto, Vortex, Vertica SQL, Tez.
• Обработка потоков. Системы потоковой обработки занимают большие объемы дискретных точек данных и выполняют непрерывный запрос для получения результатов, близких к реальному времени, по мере поступления новых данных в систему. Эта модель подходит, если у вас уже есть цифровые данные, которые постоянно генерируются (например, это мониторинг общественного настроя по проблеме, событию или продукту в социальных сетях, отслеживание новых тенденций или мониторинг логов сервера). Программы, которые могут сделать это для Hadoop: Spark Stream, Storm.
• Обработка графов. Графовые алгоритмы обычно описывают связи между вершинами, которые выражаются ребрами. При передаче через MapReduce 1.x эта модель требовала много лишних накладных расходов. Эта модель подходит для отображения нелинейных отношений между вещами: это друзья в Facebook, подписчики в Twitter, распределенные графовые базы данных, которые лежат в основе социальных сетей. Программы, которые могут сделать это для Hadoop: Apache Giraph, Apache Spark’s GraphX, Hama, Titan.

Это лишь несколько альтернативных моделей и инструментов обработки. Подробное руководство по экосистеме Hadoop можно найти здесь.

4: Высокая доступность ResourceManager

В первом релизе YARN было свое узкое место: ResourceManager. Единственный JobTracker в MapReduce 1.x обрабатывал управление ресурсами, планирование задач и мониторинг работы. Ранние релизы YARN улучшили это, разделив обязанности между глобальным ResourceManager и ApplicationMaster для каждого приложения. ResourceManager отслеживал ресурсы кластера и планировал приложения, такие как MapReduce Jobs, но был единственной точкой отказа до версии 2.4, в которой была представлена архитектура Active/Standby.

В Hadoop 2.4 единый ResourceManager был заменен одним активным ResourceManager и одним или несколькими резервными. В случае сбоя активного ResourceManager администраторы могут вручную активировать один из менеджеров. Чтобы обеспечить автоматический переход на другой ресурс, можно добавить в свой стек Apache Zookeeper. Помимо прочих обязанностей по координации задач, Zookeeper может отслеживать состояние нод YARN и в случае сбоя автоматически запускать переход в режим ожидания.

Читайте также:

• Установка Hadoop в автономном режиме в Ubuntu 16.04
• Введение в большие данные: основные понятия и термины