Intoduction to Hadoop Distributed File System

Введение в распределенную файловую систему Hadoop

Вольный перевод статьи "The Hadoop Distributed File System" by K. Shvachko, H. Kuang, S. Radia, R. Chansler (http://developer.yahoo.net/blogs/hadoop/2010/05/scalability_of_the _hadoop_dist.html).

Введение

The Hadoop Distributed File System (HDFS) спроектирована для надежного хранения очень больших объемов данных и для обеспечения высокой пропускной способности работы пользовательских приложений с этими данными.


Краткий список терминов
HDFS - Hadoop Distributed File System, распределенная файловая система Hadoop.
NameNode - основная нода кластера, которая координирует запросы клиента на чтение и запись на ноды типа DataNode, кроме того выполняет ряд критических административных и кофигурационных функций. У каждого кластера может быть одна и только одна NameNode.
DataNode - нода кластера, на которой хранятся данные, кроме того выполняет ряд специфических функций, таких как проверка целостности данных, репликация данных и проч.
Блок - какая-то часть файла, с которой происходит работа как с неделимой сущностью. Блоки одного файла могут храниться на разных DataNode.
Реплкикация - создание копии данных.
Реплика - копия данных, обычно говориться в контексте копии блока файла на какой-то DataNode.
HDFS клиент - процесс, который по требованию клиентского приложения производит чтение и запись данных в HDFS.

I Арихтектура

A. NameNode
Простарнство имен (namespace) HDFS - это иерархия файлов и директорий. Файлы и директории представлены в NameNode при помощи т. н. inodes, которые хранят в себе такие атрибуты, как права, время модификации и доступа, пространство имен и дисковые квоты. Содержимое файлов разбито на большие блоки (обычно - 128 Мб (в других источниках - 64 Мб), но пользователь может самостоятельно задать размер блока для каждого файла). Каждый блок независимо реплицируется на несколько DataNodes (обычно - 3, но пользователь может сам задать этот параметр для каждого файла). NameNode поддерживает дерево имен и отображение блоков файлов на различные DataNodes: NameNode знает где лежат те или иные реплики блоков заданного файла. Когда HDFS клиент хочет прочитать файл, он прежде всего запрашивает у NameNode место расположение блоков, в которых содержится информация этого файла. NameNode возвращает ближайшие к клиенту DataNode, откуда клиент читает блоки. При записи файла, клиент простит NameNode выделить три DataNode для записи репкликаций файла. Затем клиент пишет файл в один канал для все выделенных DataNodes. Для каждого кластера существует только одна DataNode и могут существовать десятки тысяч HDFS клиентов, запросы которых выполняются параллельно.

HDFS хранит пространство имен в оперативной памяти. Данные inode, cписок блоков, относящихся к файлам, метаинформация на NameNode называется образом (image). Слепок этого образа, записанный на жесткий, называется контрольной точкой (checkpoin). Кроме того на NameNode ведется журналирование изменений, которое пишется в нативную (не HDFS) файловую систему. Для повышения надежности дополнительные копии контрольных точек и журналов (journal) могут быть расположены на других серверах. Местанахождение реплик блоков может меняться с течение времени и поэтому не является частью контрольных точек.

B. DataNodes
Каждый блок реплики на DataNode представлен двумя файлами в нативной файловой системе. Первый содержит данные как таковые, второй содержит метаданные, включающий контрольные суммы (checksums) и вреимя создания (generation stamp) для каждого блока. Размер блока на диске занимает столько, сколько в нем информации и не округляется до 64 Мб., т.е. есть в блоке всего 32Мб., то всего 32 Мб. на диске и занято.

Когда нода запускается, она соединяется (handshake) с NameNode, в это время проверяется идентификатор пространства имен (namespace ID) ноды и версия ПО. Если что-то не сходится, NameNode автоматически выключает эту DataNode. Идентификатор пространства имен выдется нодам при конфигурировании кластера. Ноды с различными идентификаторами пространства имен не пускаются в кластер. Один идентификатор пространства имен - один кластер. Новые ноды, которые присоединяются к кластеру, когда тот уже поднят, запрашивают у него этот идентификатор.

После успешной проверки на соответствие идентификатора пространства имен и версии ПО, нода регистрируется у NameNode. Нода записывает свой личный идентификатор хранилища (storage ID) (непонятно - куда, у себя или она уже знает свой и пишет в NameNode), он является внутренним идентификатором ноды. Это нужно для того, чтобы если при перезагрузке нода получит другой IP, NameNoda ее точно идентифицировала. Идентификатор хранилища, когда он зарегистрирован в NameNode, не может больше изменяться.

DataNode посылает NameNode отчеты (block reports), для идентификации реплик. Отчет содержит в себе идентфикатор блока (block id) и время создания блока (generation stamp). Первый отчет посылается немедленно, сразу после регистрации. Последующие отчеты отсылаются каждый час.

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

NameNode не обращается напрямую к DataNode. Она использует ответы на heartbeats и в них посылает инструкции, которые могут содержать команды:
- реплицировать блоки в другую ноду;
- уничтожить локальные реплики;
- перерегистрация или выключение;
- немедленный отчет.

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



D. Image and Journal
Образ пространства имен - это метаданные файловой системы, которые описывают организацию данных, таких как файлы и директории. Постоянная (persistens) запись образа на диск называется контрольной точкой (checkpoint). Журнал (journal) это лог, который пишет в начала файла те изменения, которые должны быть постоянными (persistens). Для каждой транзации, инициированной клиентом, все изменения записываются в журнал, и файл журнала синхронизируется и принудительно записывается на диск (flush-and-sync) перед тем, как изменения будут применены к HDFS.

Файлы контрольных точек никогда не изменяются NameNode, за исключением ряда случаев. Содержимое файла контрольной точки может быть измененно, когда создана новая контрольная точка после рестарта ноды, по запросу администратора или по запросу CheckpointNode. Во время стартапа NameNode инициализирует образ пространства имен (namespace image) из контрольной точки и затем повторяет все изменения, которые есть в журнале, но которые не были записаны на диск в виде контрольной точки. Новая контрольная точка и пустой журнал пишутся обратно на диск, и только потом NameNode начинает обслуживать клиентов.

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

NameNode - это многопоточная система, и запросы поступают одновременно от множества клиентов. Сохранение транзакций на диск - это узкое место системы, в том с смысле, что все остальные потоки вынуждены ждать пока пока пройдет процедура принудительной записи на диск и синхронизации (flush-and-sync) для одного запроса. Для оптимизации процесса, NameNode разбивает запросы от различных клиентов на группы, когда запускается процесс сохранения и синхронизации одного запроса, этот процесс запускается и для всех запросов из этой группы (для всех действий, который уже были выполнены на данный момент). Остальным потокам, кроме инициатора достаточно будет только проверить, что транзации сохранены, и не нужно будет запускать процесс flush-and-sync.

E. CheckpointNode
NameNode в HDFS в дополнение к своей главной роли - обслуживания клиентских запросов - может исполнять, как альтернативу, ище две роли - CheckpointNode или BackupNode. Эти дополнительные роли специфицируются на этапе старта NameNode.
CheckpointNode время от времени комбинирует существующие контрольные точки и журналы, для создания новых контрольных точек и пустого журнала. CheckpointNode обычно запускается на другом хосте, потому что CheckpointNode требуется столько же памяти, сколько и NameNode. CheckpointNode запрашивает контрольные точки и журналы у NameNode, мержит их локально, и затем отсылает новую контрольную точку NameNode. Переодическое создание контрольных точек - это один из способов для сохранения метаданных системы. Система может стартоват с последней контрольной точки, если все резервные копии и журналы недоступны.
Когда NameNode получает новую контрольную точку, она обрезает хвост журнала, до того момента, который вошел в эту контрольную точку. HDFS кластер может работать достаточно долго, чтобы за это время очень сильно разросся журнал. Если журнал очень большой, вероятность потери или порчи данных очень велика. Кроме того, чем больше журнал, тем дольше нода стартует. Для больших кластеров нода может стартовать около часа с журналом недельной давности. Хорошая практика - делать ежедневные контрольные точки.

F. BackupNode
Также как и CheckpointNode, BackupNode способна переодически делать контрольные точки, но в дополнение к этому, может делать слепок образа пространства имен (up-to-date image) в памяти, который всегда синхронизирован с текущим состоянием NameNode. BackupNode поддерживает потоки журналов, сохраняет журналы в свои собственные хранилища, и применяет изменения к собственному образу файловой системы. Если NameNoda падает, образ в BackupNode и контрольная точка полностью воссоздадут состояние NameNode до падения. BackupNode также сама может создавать контрольные точки, запрашивая контрольные и журналы с активной NameNode, с того момента, когда NameNode проапдейтила свой образ. Это позволяет сделать процесс создания контрольных точек более эффективным. Полученная контрольная точка будет храниться на локальном диске BackupNode.

BackupNode может быть рассмотрена как read-only NameNode. Она содержит всю информацию, кроме местонахождения блоков информации. Она может выполнять все операции NameNode, за исключением операций изменения образа и возвращения информации о местонахождении блоков. Использование BackupNode позволяет запускать NameNode без постоянного хранилища, делигируя персистентность образа BackupNode.

G. Upgrades, File System Snapshots
Во время апгрейда системы, она может упасть. Падение может быть вызвано как ошибками в программном обеспечении или человеческими ошибка. HDFS позволяте делать копию состояния системы (snapshot) для минимизации потерь, если система упадет.

Администратор может сохранить на диск копию состояния системы, и в случае ошибок при апгрейде откатить систему к стабильному состоянию. Можно сделать только одну копию состояния системы в любой момент работы кластера.

Если администратор запросил создание копии состояния системы, NameNode прежде всего прочитает контрольную точку и журнал, чтобы их смержить в памяти. Затем она запишет на диск новую контрольную точку и пустой журнал в другое хранилище, при этом текущий журнал и контрольная точка отанутся неизменными. Затем, через NameNode попросит DataNodes сделать локальную копию состояния (local snapshot). У каждой DataNode будет своя копия вне зависимости от других нод. Копия состояния будет представлена в виде жестких ссылок на блоки, а не ввиде репликации всех данных. Когда DataNode удаляет блок, она лишь удаляет ссылку на него. При изменении блока, старая версия блока остается в "старой" директории.

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

Апгрейд может иметь новую структуру данных, поэтому при запуске будет перевод старых форматов в новые.

HDFS не разделят версии для NameNode и DataNode, апгрейд должен пройти весь кластер. Если во время агрейда будет испорчен или потерян образ, то резервное копирование только пространства имен приведет к полной потере данных, т.к. NameNode не будет признавать отчеты от DataNode и будет просить удалить блоки. Откат в данном случае приведет к восстановлению метаданных, но данные как таковые будут потеряны. Т. о. копии состояния нужны для того, чтобы избежать катастрофических последствий.


II Чтение и запись файлов. Менеджмент репликаций

A. File Read and Write
Приложение добавляет данные в HDFS путем создания нового файла и записи данных в него. После того, как файл закрывается, записанные байты не могут быть изменены или удалены, кроме как в случае, когда файл будет открыт и к нему буду дописаны новые данные. HDFS построена на модели “single-writer, multiple-reader”. HDFS клиент, который хочет записать в файл, арендует (lease) его. Пишущий клиент время от времени продлевает аренду, посылая NameNode информацию об этом в heartbeats. Когда файл закрывается, аренда аннулируется. Время аренды связано с гибким лимитом (soft limit) времени и жестким лимитом (hard limit) времени. Пока не закончится гибкий лимит, писателю предоставляетс эксклюзивный доступ к файлу. Если лимит закончился и писатель не успел закрыть файл или взять новую аренду, другой писатель может опередить текущего писателя. После окончания жесткого лимита (1 час), в случае, если клиент не взял аренду заново, HDFS считает, что клиент ушел и автоматически закрывает файл от имени клиента и восстанавливает аренду. Аренда файла позволяет только одному писателю писать в него. Читателей может быть много.

HDFS файл состоит из блоков. Когда файлу нужен новый блок, NameNode выделяет блок с уникальным идентификатором (block ID) и определяет список DataNode, на которых буду располагаться реплики блока. DataNodes формируют канал (pipeline) в таком порядке, чтобы минимизировать суммарное расстояние по сети от клиента до последней DataNode в этом канале. Байты будут отправлены в этот канал как последовательность пакетов. Байты, который были отправлены клиентом на клиентской стороне буферизируются, (обычно - 64 Кб). Следующий пакет будет отправлен только после того, как придет подтверждение о доставке от всех предыдущих пакетов. Количество выданных неподтвержденных пакетов ограничивается размером окна на стороне клиента.

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



t0 --> t1 - подготовительный этап;
t1 --> t2 - этап передачи данных, t1 - время отсылки первого пакета, t2 - время получения последнего подтверждения о доставке;
t2 --> t3 - завершающий этап.

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

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

HDFS позволяет читать файл, когда он открыт для записи. Когда читаемый файл открыт для записи, длина последнего блока записывается как неизвестная в NameNode. В этом случае клиент запрашивает одну из реплик с самым последним по времени значением длины до того, как начать читать содержимое.

B. Block Placement
Для больших кластеров может быть не практичным соединение всех нод в плоской топологии. Общепризнанной практикой является расположение нод на нескольких стойках. Ноды на стойках общий общий свитч, а эти свитчи в свою очередь связаны через базовые свтичи. Соединение между двумя нодами на различных стойках идет через несколько свитчей. В большинстве случаев, пропускная способность сети между нодами одной и той же стойки выше, чем пропускная способность между нодами на разных стойках.



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

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

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

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

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


C. Replication management
NameNode постоянно стремится убедиться, что блок имеет необходимое количество реплик. NameNode находит ноды, которые недореплицированы (недостаточное количество реплик) или перереплицированы (избыточное количество реплик), когда получает отчеты от DataNode. Когда блок перереплецирован, NameNode выбирает реплику, которую необходимо удалить. NameNode старается не уменьшать количество стоек, на которых расположены реплики, а также выбирать ноды с наименьшим количеством свободного пространства. Преимущество - в балансировке хранилища без уменьшения доступности блоков.

Если блок является недореплецированным, NameNode помещает его в очередь репликации. Эта очередь является очередью с приоритетами. Блок с одной репликой имеет наивысший приоритет, блок который реплецирован на две трети от необходимого имеет нименьший приоритет. Фоновый процесс периодически считывает начала очереди, для того чтобы решить, где расположить следующую реплику. Если количестов реплик блока - одна, HDFS помещает новую реплику блока в другу стойку. Если блок имеет две реплики, и если две существующие реплики располагаются в одной стойке, реплика помещается в другую стойку; в противном случа, реплика помещается на другую ноду в этой же стойке. Выгода в стоимости создания новых реплик.

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

D. Balancer
Balancer - это средство для балансировки дискового пространства в HDFS кластере. Он использует пороговое значение, которое представляется в виде дроби и лежит в промежутке (0, 1). Кластер считается сбалансированным, если для каждой DataNode ее использование (соотношение использованного пространства ко всему пространству) отличается от среднего использования по всему кластеру не более чем на пороговое значение.

Это средство установлено как приложение, которое может быть запущено администратором. Оно итеративно перемещает реплики с нод с высоким уровнем использования в ноды с низким уровнем использования дискового пространства. Balancer пользуется только одним условием - доступность данных. Когда реплика перемещается, Balancer гарантирует, что перемещение не вызовет уменьшения количества реплик и уменьшения количества стоек, в которых эти реплики расположены.

Balancer оптимизирует процесс балансировки путем минимизации копирования данных между стойками. Если Balancer решает, что реплику A необходимо переместить в другую стойку, которая в свою очередь имеет репклику B того же блока, то данные будут скопированы из реплики B а не реплики A.

Второй параметр конфигурации Balancer - это предел пропускной способности при операциях перемещения реплик. Чем он выше, тем быстрее кластер сбалансируется, но тем выше будет конкуренция с пользовательскими процессами.

E. Block Scanner
У каждой DataNode существует сканер блоков, который периодически сканирует ее реплики блоков и верифицирует контрольные суммы. Если клиент читает полный блок и проверка конрольной суммы завершается удачно, он информирует NameNode и данная DataNode рассматривает это сообщение как проверку реплики.

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

F. Decommissioing
Администратор кластера специфицирует, какие ноды могут присоединиться к кластеру, а какие нет, т.е. каким хост-адресам разрешено присоединения, а каким хост-адерсам запрещено. Администратор может дать команду переконфигурации разрешенных и запрещенных адресов. Существующий член кластера, который из него исключается маркируется как находящаяся в процессе декомпозиции. Такие ноды больше не могут быть выбраны в качестве новго хранилища блоков, однако они по прежнему обслуживают запросы на чтение. NameNode запускает процесс репликации данных с этой ноды на другие. После этого нода в NameNode маркируется как прошедшая декомпозицию и может быть безопасно отсоединена.