Распределенная обработка данных с помощью Hadoop: Часть 2. Двигаемся дальше

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

Распределенная архитектура Hadoop

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

В первой статье Hadoop был сконфигурирован таким образом, что все его демоны работали на одной машине. Итак, сначала выясним, как можно распределить Hadoop для выполнения параллельных операций. В распределенной конфигурации Hadoop имеется главный (master) узел и несколько подчиненных (slave) узлов (рисунок 1).

Как показано на рисунке 1, на главном узле работают демоны namenode, secondarynamenode и jobtracker (так называемые master-демоны). Кроме того, с этого узла осуществляется управление кластером (с помощью утилиты Hadoop и Web-браузера). На подчиненных узлах работают демоны tasktracker и datanode (slave-демоны). Отличие данной конфигурации заключается в том, что на главном узле работают демоны, отвечающие за управление и координирование кластера Hadoop, тогда как на подчиненных узлах работают демоны, обеспечивающие функции хранения данных в файловой системе HDFS и реализующие функционал MapReduce (функция обработки данных).

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

Рисунок 2. Конфигурация кластера Hadoop
 

Чтобы упростить развертывание, будем использовать виртуализацию, которая позволяет получить некоторые преимущества. Несмотря на небольшое снижение производительности, виртуализация позволяет создать конфигурацию Hadoop и затем клонировать ее на другие узлы. Исходя из этого, ваш кластер Hadoop должен выглядеть так, как показано на рисунке 3: на компьютере установлен гипервизор, управляющий тремя виртуальными машинами - на одной из них развернут главный узел Hadoop, а на двух других - подчиненные узлы.

Обновление конфигурации Hadoop

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

В псевдораспределенной конфигурации мы ничего не настраивали, и все работало на одном хосте с настройками по умолчанию. Сейчас необходимо обновить эту конфигурацию. Прежде всего, проверим текущую конфигурацию с помощью команды update-alternatives, как показано в листинге 1. Эта команда сообщает нам, что конфигурация использует файл conf.pseudo (наивысший приоритет).

	
$ update-alternatives --display hadoop-0.20-conf
hadoop-0.20-conf - status is auto.
 link currently points to /etc/hadoop-0.20/conf.pseudo
/etc/hadoop-0.20/conf.empty - priority 10
/etc/hadoop-0.20/conf.pseudo - priority 30
Current `best' version is /etc/hadoop-0.20/conf.pseudo.
$ 

Далее, создадим новую конфигурацию, сделав копию файла уже существующей конфигурации (в нашем примере это файл conf.empty), как показано в листинге 1:

$ sudo cp -r /etc/hadoop-0.20/conf.empty /etc/hadoop-0.20/conf.dist
$ 

В завершение активизируем и проверим новую конфигурацию.

	
$ sudo update-alternatives --install /etc/hadoop-0.20/conf hadoop-0.20-conf   /etc/hadoop-0.20/conf.dist 40
$ update-alternatives --display hadoop-0.20-conf
hadoop-0.20-conf - status is auto.
 link currently points to /etc/hadoop-0.20/conf.dist
/etc/hadoop-0.20/conf.empty - priority 10
/etc/hadoop-0.20/conf.pseudo - priority 30
/etc/hadoop-0.20/conf.dist - priority 40
Current `best' version is /etc/hadoop-0.20/conf.dist.
$ 

Мы получили новый конфигурационный файл conf.dist, на основе которого будем создавать новую распределенную конфигурацию. На данном этапе создайте два клона этой виртуальной машины - они будут выступать в качестве узлов DataNode.

Конфигурация Hadoop для распределенной работы

Следующий шаг заключается в настройке взаимосвязей между узлами. Для этого используются файлы с именами masters и slaves, расположенные в директории /etc/hadoop-0.20/conf.dist. В нашем примере (листинг 3) имеются три узла, которым в файле /etc/hosts назначены статические IP-адреса:

	
master 192.168.108.133
slave1 192.168.108.134
slave2 192.168.108.135

Итак, на главном узле обновите файл /etc/hadoop-0.20/conf.dist/masters, обозначив главный узел, как:

master

На подчиненных узлах обновите файл /etc/hadoop-0.20/conf.dist/slaves, добавив следующие две строки:

slave1
slave2

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

Продолжим настройку конфигурации Hadoop в директории /etc/hadoop-0.20/conf.dist. В соответствии с документацией Hadoop, необходимо выполнить на всех узлах (на главном и на обоих подчиненных) следующие действия. Во-первых, необходимо указать хозяина HDFS в файле core-site.xml (листинг 4), задав адрес и порт службы namenode (заметьте, что используется IP-адрес главного узла). В файле core-site.xml задаются основные параметры Hadoop.

	
<configuration>

  <property>
    <name>fs.default.name<name>
    <value>hdfs://master:54310<value>
    <description>The name and URI of the default FS.</description>
  <property>

<configuration>

Во-вторых, необходимо задать параметры службы MapReduce jobtracker. Демон jobtracker может работать на своем собственном узле, но в этой конфигурации мы поместим его на главный узел, как показано в листинге 5. В файле mapred-site.xml задаются параметры функционала MapReduce.

	
<configuration>

  <property>
    <name>mapred.job.tracker<name>
    <value>master:54311<value>
    <description>Map Reduce jobtracker<description>
  <property>

<configuration>

И последнее, что необходимо сделать - задать фактор репликации по умолчанию (листинг 6). Этот параметр определяет количество реплик, которые будут созданы, и обычно его значение не превышает 3. В нашем примере зададим значение, равное 2 (количество узлов типа DataNode). Это значение определяется в файле hdfs-site.xml, содержащем параметры файловой системы HDFS.

	
<configuration>

  <property>
    <name>dfs.replication<name>
    <value>2<value>
    <description>Default block replication<description>
  <property>

<configuration>

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

После того как вы создали конфигурацию, необходимо отформатировать узел namenode (главный узел HDFS). Для этого используйте утилиту hadoop-0.20, указав узел namenode и требуемую операцию (-format), как показано в листинге 7.

	
user@master:~# sudo su -
root@master:~# hadoop-0.20 namenode -format
10/05/11 18:39:58 INFO namenode.NameNode: STARTUP_MSG: 
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG:   host = master/127.0.1.1
STARTUP_MSG:   args = [-format]
STARTUP_MSG:   version = 0.20.2+228
STARTUP_MSG:   build =  -r cfc3233ece0769b11af9add328261295aaf4d1ad; 
************************************************************/
10/05/11 18:39:59 INFO namenode.FSNamesystem: fsOwner=root,root
10/05/11 18:39:59 INFO namenode.FSNamesystem: supergroup=supergroup
10/05/11 18:39:59 INFO namenode.FSNamesystem: isPermissionEnabled=true
10/05/11 18:39:59 INFO common.Storage: Image file of size 94 saved in 0 seconds.
10/05/11 18:39:59 INFO common.Storage: 
  Storage directory /tmp/hadoop-root/dfs/name has been successfully formatted.
10/05/11 18:39:59 INFO namenode.NameNode: SHUTDOWN_MSG: 
/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at master/127.0.1.1
************************************************************/
root@master:~# 

После завершения форматирования можно запускать демоны Hadoop. Это делается точно так же, как и в случае псевдораспределенной конфигурации (листинг 8). Обратите внимание на то, что на этом узле запускаются демоны namenode и secondarynamenode (как видно из вывода команды jps):

	
root@master:~# /usr/lib/hadoop-0.20/bin/start-dfs.sh
starting namenode, logging to 
  /usr/lib/hadoop-0.20/bin/../logs/hadoop-root-namenode-mtj-desktop.out
192.168.108.135: starting datanode, logging to 
  /usr/lib/hadoop-0.20/bin/../logs/hadoop-root-datanode-mtj-desktop.out
192.168.108.134: starting datanode, logging to 
  /usr/lib/hadoop-0.20/bin/../logs/hadoop-root-datanode-mtj-desktop.out
192.168.108.133: starting secondarynamenode, 
  logging to /usr/lib/hadoop-0.20/logs/hadoop-root-secondarynamenode-mtj-desktop.out
root@master:~# jps
7367 NameNode
7618 Jps
7522 SecondaryNameNode
root@master:~# 

Если вы запустите команду jps на подчиненных узлах (узлы типа DataNode), вы увидите, что на каждом из них запущен демон datanode (листинг 9).

	
root@slave1:~# jps
10562 Jps
10451 DataNode
root@slave1:~# 

Следующий шаг - запуск демонов MapReduce (jobtracker и tasktracker). Команды для запуска приведены в листинге 10. Обратите внимание на то, что в результате выполнения сценария демон jobtracker запускается на главном узле (как это было определено в конфигурации, см. листинг 5), а демоны tasktracker - на каждом из подчиненных узлов. Выполнив команду jps на главном узле, мы увидим, что демон jobtracker теперь запущен (листинг 10).

	
root@master:~# /usr/lib/hadoop-0.20/bin/start-mapred.sh
starting jobtracker, logging to 
  /usr/lib/hadoop-0.20/logs/hadoop-root-jobtracker-mtj-desktop.out
192.168.108.134: starting tasktracker, logging to 
  /usr/lib/hadoop-0.20/bin/../logs/hadoop-root-tasktracker-mtj-desktop.out
192.168.108.135: starting tasktracker, logging to 
  /usr/lib/hadoop-0.20/bin/../logs/hadoop-root-tasktracker-mtj-desktop.out
root@master:~# jps
7367 NameNode
7842 JobTracker
7938 Jps
7522 SecondaryNameNode
root@master:~# 

Наконец, проверим с помощью команды jps подчиненные узлы. Здесь мы видим, что на каждом из них к демону datanode добавился демон tasktracker (листинг 11).

	
root@slave1:~# jps
7785 DataNode
8114 Jps
7991 TaskTracker
root@slave1:~# 

Взаимосвязь между сценариями запуска, узлами и демонами показана на рисунке 4. Как видно из рисунка, сценарий start-dfs запускает на соответствующих узлах демоны namenode и datanode, а сценарий start-mapred - демоны jobtracker и tasktracker.

Тестирование HDFS

Теперь, когда Hadoop запущен и работает в кластере, можно выполнить несколько тестов и убедиться, что все работает без ошибок (листинг 12). Сначала выполним команду fs с параметром df в контексте утилиты hadoop-0.20. Так же как и в ОС Linux®, эта команда просто показывает занятое и свободное дисковое пространство на выбранном устройстве. В новой, только что отформатированной файловой системе все пространство свободно. Далее, выполним команду ls для корневой директории HDFS, создадим поддиректорию, посмотрим ее содержимое и затем удалим ее. После этого проверим файловую систему с помощью команды fsck, запущенной в контексте утилиты hadoop-0.20. Результаты выполнения всех этих команд говорят о том, что файловая система в порядке. Кроме того, мы получили дополнительную информацию о том, что были обнаружены два узла datanode.

	
root@master:~# hadoop-0.20 fs -df
File system		Size	Used	Avail		Use%
/		16078839808	73728	3490967552	0%
root@master:~# hadoop-0.20 fs -ls /
Found 1 items
drwxr-xr-x   - root supergroup          0 2010-05-12 12:16 /tmp
root@master:~# hadoop-0.20 fs -mkdir test
root@master:~# hadoop-0.20 fs -ls test
root@master:~# hadoop-0.20 fs -rmr test
Deleted hdfs://192.168.108.133:54310/user/root/test
root@master:~# hadoop-0.20 fsck /
.Status: HEALTHY
 Total size:	4 B
 Total dirs:	6
 Total files:	1
 Total blocks (validated):	1 (avg. block size 4 B)
 Minimally replicated blocks:	1 (100.0 %)
 Over-replicated blocks:	0 (0.0 %)
 Under-replicated blocks:	0 (0.0 %)
 Mis-replicated blocks:		0 (0.0 %)
 Default replication factor:	2
 Average block replication:	2.0
 Corrupt blocks:		0
 Missing replicas:		0 (0.0 %)
 Number of data-nodes:		2
 Number of racks:		1

The filesystem under path '/' is HEALTHY
root@master:~# 

Выполнение задания MapReduce

Запустим задание MapReduce и убедимся, что вся конфигурация работает должным образом (листинг 13). Первое, с чего следует начать - это предоставить системе какие-то данные. Итак, давайте начнем с создания директории, в которой будут храниться наши входные данные (назовем эту директорию input). Для этого воспользуемся командой mkdir утилиты hadoop-0.20. После этого с помощью команды put поместим в файловую систему HDFS два файла. Содержимое директории input можно проверить с помощью команды ls.

	
root@master:~# hadoop-0.20 fs -mkdir input
root@master:~# hadoop-0.20 fs -put   /usr/src/linux-source-2.6.27/Doc*/memory-barriers.txt input
root@master:~# hadoop-0.20 fs -put   /usr/src/linux-source-2.6.27/Doc*/rt-mutex-design.txt input
root@master:~# hadoop-0.20 fs -ls input
Found 2 items
-rw-r--r--  2 root supergroup  78031 2010-05-12 14:16 /user/root/input/memory-barriers.txt
-rw-r--r--  2 root supergroup  33567 2010-05-12 14:16 /user/root/input/rt-mutex-design.txt
root@master:~# 

Теперь передадим на выполнение функции MapReduce задачу wordcount (листинг 14). Как и в псевдораспределенной модели, необходимо указать входную директорию input (содержащую входные файлы) и выходную директорию output (эта директория не существует, но она будет создана и заполнена данными в результате работы узла namenode).

	
root@master:~# hadoop-0.20 jar   /usr/lib/hadoop-0.20/hadoop-0.20.2+228-examples.jar wordcount input output
10/05/12 19:04:37 INFO input.FileInputFormat: Total input paths to process : 2
10/05/12 19:04:38 INFO mapred.JobClient: Running job: job_201005121900_0001
10/05/12 19:04:39 INFO mapred.JobClient:  map 0% reduce 0%
10/05/12 19:04:59 INFO mapred.JobClient:  map 50% reduce 0%
10/05/12 19:05:08 INFO mapred.JobClient:  map 100% reduce 16%
10/05/12 19:05:17 INFO mapred.JobClient:  map 100% reduce 100%
10/05/12 19:05:19 INFO mapred.JobClient: Job complete: job_201005121900_0001
10/05/12 19:05:19 INFO mapred.JobClient: Counters: 17
10/05/12 19:05:19 INFO mapred.JobClient:   Job Counters 
10/05/12 19:05:19 INFO mapred.JobClient:     Launched reduce tasks=1
10/05/12 19:05:19 INFO mapred.JobClient:     Launched map tasks=2
10/05/12 19:05:19 INFO mapred.JobClient:     Data-local map tasks=2
10/05/12 19:05:19 INFO mapred.JobClient:   FileSystemCounters
10/05/12 19:05:19 INFO mapred.JobClient:     FILE_BYTES_READ=47556
10/05/12 19:05:19 INFO mapred.JobClient:     HDFS_BYTES_READ=111598
10/05/12 19:05:19 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=95182
10/05/12 19:05:19 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=30949
10/05/12 19:05:19 INFO mapred.JobClient:   Map-Reduce Framework
10/05/12 19:05:19 INFO mapred.JobClient:     Reduce input groups=2974
10/05/12 19:05:19 INFO mapred.JobClient:     Combine output records=3381
10/05/12 19:05:19 INFO mapred.JobClient:     Map input records=2937
10/05/12 19:05:19 INFO mapred.JobClient:     Reduce shuffle bytes=47562
10/05/12 19:05:19 INFO mapred.JobClient:     Reduce output records=2974
10/05/12 19:05:19 INFO mapred.JobClient:     Spilled Records=6762
10/05/12 19:05:19 INFO mapred.JobClient:     Map output bytes=168718
10/05/12 19:05:19 INFO mapred.JobClient:     Combine input records=17457
10/05/12 19:05:19 INFO mapred.JobClient:     Map output records=17457
10/05/12 19:05:19 INFO mapred.JobClient:     Reduce input records=3381
root@master:~#

Последнее, что нужно сделать - проанализировать полученные результаты. Поскольку задача wordcount обрабатывалась с помощью функции MapReduce, результатом будет являться один файл, собранный в единое целое из отдельных параллельно обработанных файлов. Этот результирующий файл содержит строки, каждая из которых содержит слово, найденное во всех входных файлах, и число, показывающее, сколько раз это слово было найдено (листинг 15).

	
root@master:~# hadoop-0.20 fs -ls output
Found 2 items
drwxr-xr-x   - root supergroup          0 2010-05-12 19:04 /user/root/output/_logs
-rw-r--r--   2 root supergroup      30949 2010-05-12 19:05 /user/root/output/part-r-00000
root@master:~# hadoop-0.20 fs -cat output/part-r-00000 / head -13
!=		1
"Atomic		2
"Cache		2
"Control		1
"Examples		1
"Has		7
"Inter-CPU	1
"LOAD		1
"LOCK"		1
"Locking		1
"Locks		1
"MMIO		1
"Pending		5
root@master:~# 

Web-интерфейсы управления

Несмотря на универсальность и богатый функционал утилиты hadoop-0.20, иногда удобнее использовать графический интерфейс. Вы можете подключиться в Web-браузере к узлу namenode по адресу http://master:50070 и к узлу jobtracker по адресу http://master:50030. Через интерфейс namenode можно получать информацию о файловой системе HDFS, как показано на рисунке 5 (здесь мы просматриваем информацию о директории input).

Через интерфейс jobtracker можно получать информацию о выполняющихся или завершенных заданиях. На рисунке 6 показана информация о вашем последнем задании (см. листинг 14), полученная в результате выполнения запроса к JAR-архиву, а также статус и количество заданий. Обратите внимание на то, что были выполнены две операции map (по одной для каждого входного файла) и одна операция reduce (чтобы собрать два фрагмента, на которые был разбит входной файл).

Наконец, через интерфейс namenode вы можете проверить статус всех узлов DataNode. На главной странице namenode показано количество работающих и неработающих узлов. Для каждого узла присутствует гиперссылка, позволяющая получить более подробную информацию. На рисунке 7 показаны работающие узлы типа DataNode и статистика для каждого из них.

С помощью Web-интерфейсов namenode и jobtracker можно получить много другой информации - здесь приведен лишь небольшой пример. На этих Web-страницах вы найдете большое количество гиперссылок, с помощью которых сможете получить дополнительную информацию о конфигурации и работе Hadoop (включая журналы времени выполнения).

Из этой статьи вы узнали, как из псевдораспределенной конфигурации Hadoop от компании Cloudera можно получить полностью распределенную конфигурацию. Крайне малое количество требуемых действий, а также идентичный интерфейс для приложений MapReduce делают Hadoop исключительно полезным для распределенной обработки. Также интересно посмотреть на схему масштабирования Hadoop. Добавляя новые узлы DataNode, вы можете легко масштабировать Hadoop до еще более высокого уровня распределенной обработки. В третьей, последней части этой серии статей мы рассмотрим процесс создания приложения MapReduce для Hadoop.

Ссылки по теме

• Подробнее о решениях IBM для управления большими данными (IBM Big Data)
• Статьи по теме Big Data
• Новости по теме Big Data
• Услуги по внедрению решений для обработки больших объемов данных
• Обратиться в Интернет-магазин ITshop.ru за дополнительной информацией/по вопросу приобретения продуктов
• Консультация по выбору решений и технологий для обработки больших объемов данных