Apache Hadoop Tutorial — ULTIMATE Guide (PDF Download)

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

Хотя возможно установить Apache Hadoop в операционной системе Windows, GNU / Linux является базовой платформой для разработки и производства. Для установки Apache Hadoop необходимо выполнить следующие два требования:

• Java> = 1.7 должна быть установлена.
• ssh должен быть установлен и sshd должен быть запущен.

Если ssh и sshd не установлены, это можно сделать с помощью следующих команд в Ubuntu:

Теперь, когда ssh установлен, мы создаем пользователя с именем hadoop , который позже установит и запустит кластер HDFS и задания MapReduce:

Как только пользователь создан, мы открываем для него оболочку, создаем для него пару ключей SSH, копируем содержимое открытого ключа в файл authorized_keys и проверяем, что мы можем войти в localhost, используя ssh без пароля:

Настройка Hadoop как кластера очень похожа на настройку «одного узла». В основном должны быть выполнены те же шаги, что и выше. Единственное отличие состоит в том, что кластеру нужен только один NameNode, то есть мы должны создавать и настраивать каталог для NameNode только на узле, который должен запускать экземпляр NameNode (главный узел).

Файл $HADOOP_HOME/etc/hadoop/slaves можно использовать, чтобы сообщить Hadoop обо всех машинах в кластере. Просто введите имя каждого компьютера в виде отдельной строки в этом файле, где первая строка обозначает узел, который должен быть ведущим (то есть запускает NameNode):

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

Мы изменили следующие строки в файле $HADOOP_HOME/etc/hadoop/core-site.xml , чтобы он содержал DNS-имя главного узла на всех машинах:

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

Если подчиненные файлы настроены правильно и доверенный доступ (т. Е. Доступ без пароля для пользователя hadoop) настроен для всех компьютеров, перечисленных в файле slaves , кластер можно запустить с помощью следующей командной строки на этом главном узле:

По желанию, вместо автоматического запуска кластера с главного узла, можно также использовать сценарий hadoop-daemon.sh для запуска либо NameNode, либо DataNode на каждой из машин:

Заменив команду start на stop , позже можно остановить выделенные серверы:

Обратите внимание, что сначала все DataNodes должны быть остановлены. NameNode останавливается после завершения работы всех узлов данных.

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

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

В HDFS существует два разных типа серверов: NameNodes и DataNodes. Хотя существует только один NameNode, количество DataNode не ограничено.

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

Тот факт, что весь кластер имеет только один NameNode, делает всю архитектуру очень простой, но также вводит одну точку отказа (SPOF). До Hadoop 2.0 не было возможности запустить два узла NameN в режиме отработки отказа, то есть, когда один узел NameN не работал из-за сбоя машины или из-за операции обслуживания, весь кластер не работал. Более новые версии Hadoop позволяют запускать два экземпляра NameNode в активной / пассивной конфигурации с горячим резервированием. В качестве требования для этой конфигурации предполагается, что на обоих компьютерах, на которых работают серверы NameNodes, должно быть одинаковое оборудование и общее хранилище (которое снова должно быть доступно).

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

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

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

Вся информация о пространстве имен HDFS хранится в NameNode и хранится в памяти. Изменения в этой структуре данных записываются как записи в журнал транзакций с именем EditLog. Этот EditLog хранится как обычный файл в файловой системе операционной системы, в которой работает NameNode. Использование журнала транзакций для записи всех изменений позволяет восстановить эти изменения в случае сбоя NameNode. Следовательно, когда NameNode запускается, он считывает журнал транзакций с локального диска и применяет все сохраненные в нем изменения к последней версии пространства имен.

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

Узлы данных хранят блоки в своей локальной файловой системе и распределяют файлы по каталогам, чтобы локальная файловая система могла эффективно с ними справляться. При запуске DataNode сканирует структуру локальной файловой системы и затем отправляет список всех блоков, которые он хранит, в NameNode (BlockReport).

HDFS устойчива к нескольким типам сбоев:

• Ошибка DataNode: каждый DataNode время от времени отправляет сообщение пульса в NameNode. Если NameNode не получает пульса в течение определенного периода времени от DataNode, узел считается мертвым, и дальнейшие операции для него не запланированы. Мертвый DataNode уменьшает коэффициент репликации сохраненных блоков данных. Чтобы предотвратить потерю данных, NameNode может запустить новые процессы репликации, чтобы увеличить коэффициент репликации для этих блоков.
• Сетевые разделы: если кластер разбивается на два или более разделов, NameNode теряет соединение с набором DataNodes. Эти узлы данных считаются мертвыми, и дальнейшие операции ввода-вывода для них не запланированы.
• Целостность данных: когда клиент загружает данные в файловую систему, он вычисляет контрольную сумму для каждого блока данных. Эта контрольная сумма хранится в скрытом файле в том же пространстве имен. Если тот же файл читается позже, клиент, считывающий блоки данных, также извлекает скрытый файл и сравнивает контрольные суммы с теми, которые он вычисляет для полученных блоков.
• Ошибка NameNode: поскольку NameNode является единственной точкой отказа, очень важно, чтобы его данные (EditLog и FsImage) могли быть восстановлены. Поэтому можно настроить хранение более одной копии файлов EditLog и FsImage. Хотя это уменьшает скорость, с которой NameNode может обрабатывать операции, одновременно обеспечивается наличие нескольких копий критических файлов.

Теперь, когда мы изучили основы инфраструктуры MapReduce, пришло время реализовать простой пример. В этом уроке мы собираемся создать «инвертированный индекс» из пары текстовых файлов. «Инвертированный индекс» означает, что мы создаем для каждого слова список текстовых файлов, в которых оно встречается. Этот вид индекса используется поисковыми системами, такими как Google, Elasticsearch или Solr, для поиска всех страниц, которые указаны для поискового запроса. В реальных условиях упорядочение списка является наиболее важной работой, так как мы ожидаем, что наиболее «релевантные» страницы будут перечислены первыми. В этом примере мы, конечно, реализуем только первый базовый шаг, который сканирует набор текстовых файлов и создает для него соответствующий «инвертированный индекс».

Мы начнем с создания проекта maven в командной строке (и предположим, что maven настроен правильно):

Это создаст следующую структуру в файловой системе:

Мы добавляем следующие строки в файл pom.xml :

Раздел dependencies определяет артефакты maven, которые мы будем использовать (здесь: библиотека hadoop-client в версии 2.7.1). Чтобы запустить наше приложение без указания основного класса в командной строке, мы определяем основной класс с помощью maven-jar-plugin .

Этот основной класс выглядит следующим образом:

После создания объекта Configuration создается новый экземпляр Job , передавая конфигурацию и имя задания. Далее мы указываем класс, который реализует задание MapReduce а также классы, которые реализуют Mapper , Combiner и Reducer . Поскольку шаг Combiner — это в основном только шаг Reduce который выполняется локально, мы выбираем тот же класс, что и для шага Reducer . Теперь необходимо задание MapReduce того типа, какой выходной ключ и значения. Поскольку мы собираемся предоставить список документов для каждого термина, мы выбираем класс Text . Наконец, мы предоставляем путь для входных и выходных документов и запускаем работу, вызывая waitForCompletion .

На шаге Map нашей реализации MapReduce мы собираемся разбить входной документ на термины и создать промежуточные пары ключ / значение в форме (термин / документ):

Реализация mapper в основном подклассирует класс Mapper и переопределяет метод map() . В этом методе значение (т. Е. Документ) разбивается на токены, и каждый токен записывается в Context как ключ. Создание токенов с помощью StringTokenizer конечно, очень простой способ, и его недостаточно для многих других типов документов, но в этом примере показано, как создавать разные ключи для одной пары ключ / значение ввода. Значением промежуточной пары ключ / значение должно быть имя файла документа, поэтому мы вызываем метод getInputSplit() в контексте, FileSplit его к FileSplit и записываем полученный путь в контекст.

Промежуточные пары ключ / значение, созданные этой реализацией Mapper , затем передаются в Reducer :

Как и Mapper , Reducer подклассирует соответствующий класс из среды Hadoop и переопределяет метод reduce() . Этот метод reduce для каждого промежуточного ключа список значений. В нашем случае это список путей документов, которые содержат термин (ключ). Поскольку этот список может быть настолько длинным, что он не помещается в память, платформа предоставляет итератор для значений. Но в этом простом примере список документов будет достаточно коротким, чтобы мы могли хранить все пути документов в LinkedList , чтобы отфильтровать дубликаты. Выходная пара ключ / значение — это просто термин с разделенным запятыми списком документов, содержащих этот термин.

После компиляции исходного кода с помощью mvn install мы можем загрузить набор документов в HDFS. Пример проекта содержит несколько текстовых документов в папке src/test/resources которые можно загрузить в HDFS с помощью следующей команды:

Это создаст новую папку с именем input в домашнем каталоге пользователя hadoop (при условии, что команда выполнена как пользователь hadoop). Если папка уже существует, ее можно удалить, вызвав:

Теперь мы можем запустить работу MapReduce:

Если работа завершится успешно, мы увидим строку, похожую на следующую в выводе:

Следующая команда сообщает нам, как называются выходные файлы:

Файл с именем _SUCCESS является только маркерным файлом и не содержит никаких данных. Фактический вывод хранится в файле part-r-00000 :

Как и ожидалось, этот файл теперь содержит упорядоченный список всех терминов вместе с путями к документам, которые содержат это слово. Так, например, термин «И» содержится в двух документах «UseJUnitsExpectedExceptionsSparingly.txt» и «MicroservicesUseCases.txt».

YARN (еще одно средство согласования ресурсов) было представлено в Hadoop с версией 2.0 и решает несколько проблем с планированием ресурсов MapReduce в версии 1.0. Чтобы понять преимущества YARN, мы должны рассмотреть, как планирование ресурсов работало в версии 1.0.

Задание MapReduce разбивается платформой на задачи (задачи Map, задачи Reducer), и каждая задача выполняется на компьютерах DataNode в кластере. Для выполнения задач каждый компьютер DataNode предоставил заранее определенное количество слотов (слотов карты, слотов редукторов). JobTracker отвечал за резервирование слотов выполнения для различных задач задания и следил за их выполнением. Если выполнение не удалось, он зарезервировал другой слот и перезапустил задачу. Он также очищает временные ресурсы и делает зарезервированный слот доступным для других задач.

Теперь, когда мы узнали об архитектуре YARN, пришло время выполнить задание MapReduce в YARN. Поэтому мы настраиваем в файле, $HADOOP_HOME/etc/hadoop/mapred-site.xmlчто должна использоваться инфраструктура YARN:

Затем мы добавляем следующее свойство в файл $HADOOP_HOME/etc/hadoop/yarn-site.xml:

Теперь пришло время запустить менеджер ресурсов:

Когда мы запускаем задание MapReduce, используя YARN, результат немного отличается:

Как показывают сообщения журнала, можно отслеживать задания, используя веб-интерфейс, работающий через порт 8088:

Это был учебник по Apache Hadoop.