Релиз Hadoop 1.0 - мифы и реальность

После шести лет разработки представлен релиз проекта Apache Hadoop 1.0.0, который базируется на стабильной ветке 0.20.203.x и включает в себя поддержку безопасного режима работы кластера (наработки ветки "0.20-security" (0.20.205.0) с поддержкой аутентификации с использованием Kerberos и средств разграничения полномочий). Официальные Release Notes.

Я долго ждал этого релиза, и не в последнюю очередь, по причине часто изменяющегося API. Сегодня, хотелось бы рассказать об особенностях этого продукта. Не буду рассказывать про "космические корабли и большой театр" (про архитектуру, общие слова, успешные истории и работу кластера), с этим неплохо справляется Иван Блинков. Мы же, остановимся на деталях и советах. И так, мифы и реальности...

Чем НЕ является Hadoop?

Это не реляционная база данных. Здесь нельзя создать таблицу, сохранить в нее данные, а затем выполнить SELECT. Возможно, это банально, прошу прощения специалистов, но об этом следует знать тем кто принимает решение о его внедрении. И пусть вас не вводит в заблуждение проект Hive, выполненный как надстройка над Hadoop (он даже имеет SQL подобный язык HiveQL). И даже Hive не является полноценной реляционной базой данных или её адекватной заменой. Вас может неприятно удивить, что простой запрос на миллионе записей может выполняться несколько минут! А как же Hbase (там же написано: "HBase is the Hadoop database. Think of it as a distributed scalable Big Data store.")? Hbase это key value хранилище.  Где тоже есть трудности с произвольными запросами (если вы не знаете ключа записи или таймстемпа). Пожалуйста НЕ путайте и остерегайтесь ошибок в принятии архитектурных решений. Hadoop - это "распределенная файловая система с бонусом". Причем, работа с этой распределенной файловой системой (HDFS) возможна только "изнутри" приложения и требует наличия соответствующих подключаемых библиотек и конфигурационных файлов. Подмонтировать HDFS раздел и использовать его как часть локальной файловой системы БЕЗ дополнительных прослоек не получится.

Бонусы, сопровождающие HDFS - это Map/Reduce framework (подробнее о нем мы поговорим ниже). На объеме данных менее нескольких десятков миллионов записей, скорее всего, вы почувствуете только множество ограничений и неудобств, используя Map/Reduce.

Hadoop - кластер, масштабируемый, защищенный от сбоев? 

Да, это так, но здесь тоже есть "нюансы". Начнем с распространенного мифа. Ресурсы ограничены, поэтому для тестов развернем кластер на виртуальных машинах! Ну, это же для тестов :) Такое вполне допустимо, но при одном условии - вы используете, хотя бы, ESXi вместо ESX или другую систему виртуализации, в которой не бывает "скачков во времени". Как это проявляется. В обычном режиме, без нагрузки ESX ведет себя вполне адекватно. Однако, когда идет интенсивный ввод/вывод (а на тестах такое случается, не говоря уже про боевой режим), могут происходить "странные" штуки. Например, изучая логи, я неоднократно сталкивался с тем, что время записи в лог выглядело примерно так:

12:32:24.126 - [Thread1] - ...
12:32:24.127 - [Thread1] - ... (!)
12:32:28.971 - [Thread1] - ...
12:32:28.975 - [Thread1] - ...
...
12:32:24.128 - [Thread1] - ... (!)
12:32:24.133 - [Thread1] - ... 
...

Вдруг, время совершает прыжок на 4-ре секунды!!! Но, совсем не на долго :) Потом, все нормализуется и виртуальная машина продолжает работать, а вместе с ней и Java приложения. Безобидная коррекция системного времени. Не каждое Java приложение, особенно работающее в кластере, считает такие "скачки во времени" безобидными. На моей практике, в большинстве случаев, это приводило к разрыву TCP/IP соединений. Не каждое Java приложение умеет восстановить целостность кластера и обеспечить когерентность данных в условиях "частичного развала" кластера. Причина понятна и не тривиальна (например, даже JGroups до сих пор не научился это делать безупречно, Hadoop не исключение). Совет - НЕ используйте Hadoop на виртуальных машинах имеющих проблемы с синхронизацией системного времени.

Теперь поговорим о размере кластера. Особенности архитектуры и протокола HDFS не позволяют иметь в кластере более 4000 хостов (Data Nodes) без существенной деградации производительности всего кластера. Поэтому - используйте "шардинг" (несколько групп кластеров) при приближении к этому порогу.

Hadoop Name node. Один из важнейших компонентов всего кластера. 

Он содержит "карту" с адресами - где и на каких дата нодах записан файл или его части. Это "центральный индекс" всего кластера. Понятно, что если его не станет, то любые операции станут невозможны. Стоит ли говорить, что если произойдет сбой диска и будет утеряна метаинформация о HDFS, весь кластер умрет? Поэтому, совет - создавайте 2-ве Name ноды, записывайте метаданные на несколько дисков и не забывайте делать их резервное копирование.   

Вы имеете 1+ Name ноду и 2+ Data ноды. Пожалуйста, убедитесь в том, что фактор репликации 3 или более соответствует реальному количеству хостов, хранящих данные!!! Если ваш кластер имеет 2-ве Name ноды и 2-ве Data ноды, то вы физически НЕ сможете разместить фрагменты одного файла в 3-х экземплярах. Совет - работоспособный Hadoop кластер начинается с 5-ти реальных (НЕ виртуальных) хостов. 

Правильно настроенный Hadoop кластер работает стабильно и не боится выхода из строя одного или более (зависит от размера кластера) хостов одновременно. Добавление новой Data ноды, происходит абсолютно "прозрачно". Советы... Используйте относительно дешевое железо под Data ноды. Храните данные в отдельном разделе файловой системы (как это часто делают для "/home"). В случае выхода из строя Data ноды, не пытайтесь её реанимировать или восстановить на ней данные (конечно если их у вас больше 3-х и фактор репликации не ниже 3), проще отформатировать раздел где они хранились и ввести в состав кластера новую Data ноду.   

"Secondary NameNode" (deprecated ?!?)

Как понятно из предыдущего раздела, необходимо записывать данные Name ноды на несколько дисков. Однако, даже запись на несколько дисков НЕ гарантирует работу кластера, в случае остановки самой Name ноды. Да, данные сохранены, даже в нескольких копиях, но в отсутствии Name ноды некому эти данные предоставлять по запросу. Теперь, для дублирования Name ноды существуют Checkpoint и Backup ноды. Пожалуйста, не забывайте об организации отказоустойчивости и правильно конфигурите свою систему! Иначе, вас ждет полное разочарование, только из-за того, что не смогли это "приготовить".

Записал и забыл (и/или потерял) 

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

Пример записи в локальный файл

// ...
OutputStream out = null; byte[] buf = new byte[1024*32]; 
try {
  out = new FileOutputStream("/tmp/file.dat");
  // loop begin
  int bytesCount = fillBuffer(buf); // заполняем буфер записи
  out.write(buf,0,bytesCount);
  // loop end
  out.flush();
} finally {
  if (out!=null) try { out.close(); } catch(Throwable err) {}
}

Пример записи в HDFS

// ...
FSDataOutputStream out = null; byte[] buf = new byte[1024*32];
FileSystem fs = null;
try {
  fs = getFileSystem(); // получаем объект для работы с HDFS
  out = fs.create(new Path("/tmp/file.dat"));
  // loop begin
  int bytesCount = fillBuffer(buf); // заполняем буфер записи
  out.write(buf,0,bytesCount);
  // loop end
  out.flush();
} finally {
  if (out!=null) try { out.sync(); } catch(Throwable err) {} // (!!!)
  if (out!=null) try { out.close(); } catch(Throwable err) {}
}

Обратите внимание, для работы с HDFS мы используем специальный класс org.apache.hadoop.fs.FileSystem, предоставляющий функции управления каталогами и файлами. В остальном, все как с обычным OutputStream, за исключением вызова out.sync(); Пожалуйста, не забывайте вызывать функции flush(), sync() и close(); при записи в HDFS.  Пренебрегая этим советом, вы можете быть неприятно удивлены, когда в файл записали, а читать нечего (нет данных) o_O. Еще, маленький нюанс! Начиная с версии 0.21 и выше flush() и sync() объявлены устаревшими. Вместо них следует использовать hflush() и hsync() соответственно.

Версия 1.0.0 - выпущена на основе 0.20.205.0 и функции flush(), sync(), close() остались прежними. Т.е. в 1.0.0 НЕ надо использовать hflush() и hsync();  

Миллионы маленьких файлов

Следующая проблема или под что НЕ "заточен" Hadoop - это хранение большого количества маленьких файлов. Если для локальной файловой системы это не проблема, могут быть лишь ограничения на количество файлов в директории, то в HDFS этого ограничения нет. Однако, каждый файл и директория имеет метаинформацию, хранящуюся на Name ноде. Представьте, что информация о 10 миллионах маленьких файлов в HDFS будет "отъедать" порядка 3G памяти Name нод! Поэтому, используйте, если это возможно, org.apache.hadoop.io.SequenceFile (пример использования)

Совет - НЕ плодите множество мелких файлов, комбинируйте их в один большой файл с индексом. Целесообразно в качестве ключа использовать абсолютный путь маленького файла, а в качестве значения его содержимое.


File Append

Одна из самых приятных штук в новом релизе - это поддержка режима добавления данных в уже записанный файл. Ведь много маленьких файлов плохо, а собрать их в несколько больших, без FileSystem::append() может быть не простой задачей.

Что бы заработал функционал добавления (по умолчанию он отключен) необходимо на серверной и клиентской стороне в файл hdfs-site.xml добавить секцию:

 
<property>
  <name>dfs.support.append</name> 
  <value>true</value>
</property>

Пример использования:

 
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);

Path f = new Path("/test/hdfs/file"+System.currentTimeMillis()+".dat"); 
FSDataOutputStream out = fs.create(f);
out.write("Превед Hadoop 1.0!!!".getBytes());
out.flush();
out.sync();
out.close();

byte[] appendedData = "\nДополнение...".getBytes(); 
out = fs.append(f,appendedData.length);
out.write(appendedData);
out.close();
  
fs.close();

(!) Обратите внимание. ::flush() и ::sync() при добавлении не вызываются. Хотя ::close() остается обязательным. Управление блокировками, dump добавленных данных, репликация и прочее отдается на откуп самому Hadopp. Мы не вмешиваемся в этот процесс. Хотя такое поведение может не устраивать вас в некоторых случаях.  

Too many open files

Ещё один "неприятный сюрприз" может вас ожидать при активном использовании I/O операций. Сюрприз в логах может проявляться как ошибка - java.io.IOException: Too many open files или java.io.EOFException или другой ошибкой, причиной которой стало превышение лимита одновременно открытых файлов. Совет - увеличить этот лимит, для пользователя от имени которого выполняется Hadoop, заданный на уровне операционной системы.


Блокировка доступа к директории/файлу

Hadoop НЕ имеет "штатных" средств блокировки. И это логично, так как у каждого приложения есть свои "тараканы". Пожалуйста, учитывайте особенности работы Hadoop кластера и сетевого взаимодействия его компонент!

Помните, что запись в "один" файл, на самом деле - это последовательное создание нескольких копий одного блока данных. Причем копии создаются на разных хостах (что позволяет, кроме надежности, улучшить характеристики чтения). Если необходимо заблокировать запись/чтение файла в одном потоке и не допустить его перезаписи в другом потоке или экземпляре приложения (на другом хосте), следует использовать внешний "координатор" (так же как и с 2-х тактными XA транзакциями). Совет - используйте ZooKeeper, либо собственную реализацию "shared memory" и глобальных мьютексов. 

Map / Reduce

Hadoop это не только распределенная файловая система (HDFS), но ещё и могучее средство исполнения Map/Reduce задач. Однако, прежде чем запускать свою задачу, жизненно необходимо её отладить! И здесь, нас ждет ещё одна "сложность". Не так просто получить доступ к отладке удаленного Java приложения, исполняющегося на реальном кластере в продуктовом окружении. Поэтому, совет - используйте несколько подходов.

Другая "трудность", с которой сталкиваются специалисты - это необходимость иметь на ВСЕХ хостах кластера JAR'ы со своими классами! Конечно, для выполнения "своего" кода требуется наличие "своих" классов. Совет - используйте несколько подходов.

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

С новым годом!

UPD: hflush(), hsync(), append mode, too many open files, secindary namenodes.