HDFS Architecture: NameNode, DataNode, блоки, репликация и heartbeat

История и философия: откуда появился HDFS¶

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

Исторический контекст: от Google к Apache¶

В 2002 году инженеры Doug Cutting и Mike Cafarella разрабатывали поисковый движок Apache Nutch. Им нужно было проиндексировать весь интернет - миллиарды страниц, - и хранить эти данные распределённо. Обычные файловые системы не справлялись: они были рассчитаны на один сервер.

В 2003 году Google опубликовала статью о Google File System (GFS) - распределённой файловой системе для хранения больших файлов на кластерах из commodity-серверов. В 2004 году Google опубликовала статью о MapReduce - модели параллельной обработки данных. Cutting и Cafarella переосмыслили эти идеи как открытый проект.

В 2006 году Hadoop был выделен из проекта Nutch в самостоятельный фреймворк. В 2008 году Yahoo выпустила Hadoop как open source. В ноябре 2012 года Apache Software Foundation сделала его публично доступным.

HDFS (Hadoop Distributed File System) - это открытая реализация идей GFS, разработанная специально для хранения очень больших файлов на кластерах из обычных (commodity) серверов с возможностью параллельного доступа для аналитических джобов.

Три ключевых цели проектирования HDFS¶

При проектировании HDFS инженеры поставили три конкретных цели, которые определили все архитектурные решения:

Цель 1: Работа с большими датасетами. HDFS рассчитан на файлы размером от гигабайт до терабайт и кластеры из сотен и тысяч узлов. Это противоположность традиционных файловых систем, оптимизированных под миллионы маленьких файлов.

Цель 2: Быстрое обнаружение ошибок и восстановление. Кластер из тысячи commodity-серверов - это норма, что один из них выйдет из строя каждую неделю. HDFS проектировался с расчётом на то, что аппаратные сбои - это обычная ситуация, а не исключение. Обнаружение сбоя и автоматическое восстановление - встроенная функциональность, а не дополнительная надстройка.

Цель 3: Перемещение вычислений к данным. Передача петабайт данных по сети между хранилищем и вычислительным кластером - экономически невыгодно и медленно. HDFS решает это через co-location: DataNode (хранилище) и вычислительные узлы (YARN NodeManager / Spark Executor) работают на одних и тех же физических серверах.

Большие данные: пять V, которые сделали HDFS необходимым¶

HDFS появился в ответ на феномен «больших данных», который характеризуется пятью измерениями - «5V»:

HDFS напрямую решает проблему Volume (петабайты данных на тысячах серверов) и Variety (хранит любые форматы: CSV, JSON, Parquet, ORC, Avro, изображения, видео - без предварительной обработки). Именно это сделало HDFS центральным элементом Big Data инфраструктуры первого поколения.

Зачем Spark-инженеру знать HDFS¶

На первый взгляд кажется, что Spark-разработчик работает на уровне DataFrame и SQL, а то, что происходит со storage - дело DevOps и инфраструктурных команд. Это заблуждение стоит дорого.

Каждый раз, когда вы вызываете spark.read.parquet("hdfs://..."), между вашим кодом и физическими данными происходит сложная хореография: Spark Driver запрашивает метаданные у NameNode, планировщик размещает таски на тех Executor, которые физически ближе к нужным блокам, а DataNode отдаёт данные напрямую в JVM процесс Executor. Если вы не понимаете этой механики, вы не сможете объяснить, почему одна джоба читает 100 GB за 30 секунд, а другая - за 10 минут при той же схеме.

Три конкретных точки влияния HDFS на производительность Spark:

1. Data Locality. Spark scheduler при планировании тасок смотрит на то, на каком узле кластера физически лежит каждый блок HDFS. Таска, запущенная прямо на том узле, где лежат данные (уровень PROCESS_LOCAL), не гоняет данные по сети вообще. Таска на соседнем узле того же rack (RACK_LOCAL) платит ~10x меньший network cost по сравнению с таской на другом rack. Если вы неправильно настроили spark.locality.wait или файлы разбиты на тысячи мелких кусков - locality ломается, и вся мощь кластера тратится на перекачку данных по сети.

2. Input Splits и начальный параллелизм. Количество начальных партиций DataFrame при чтении с HDFS напрямую определяется размером блока HDFS и параметром spark.sql.files.maxPartitionBytes (дефолт 128 MB). Если блок HDFS = 128 MB, а файл занимает 10 GB, Spark создаёт ~80 входных партиций. Понимание этой связи позволяет осознанно управлять первоначальным параллелизмом джобы без лишних repartition().

3. Проблема мелких файлов. Data Lake с миллионами мелких Parquet-файлов (каждый по 1–10 MB) убивает производительность сразу по двум каналам: каждый файл - отдельная таска в Spark (тысячи задач вместо сотен), и каждый файл - отдельная запись в памяти NameNode (риск OOM). Когда вы знаете, как HDFS считает память под метаданные, вы понимаете, почему ваш spark.write.parquet() должен завершаться компакцией результата.

Схема выше показывает ключевой принцип Hadoop: «двигай вычисления к данным, а не данные к вычислениям». Spark Driver обращается к NameNode только за метаданными (адресами блоков), а сами данные Executor читают напрямую с DataNode по TCP. Это означает, что NameNode никогда не является узким местом для data-трафика - он управляет только картой, а не перевозит груз.

Hadoop как экосистема: HDFS - это только один слой¶

Частая ошибка - отождествлять HDFS и Hadoop. HDFS является лишь одним компонентом более широкой экосистемы Apache Hadoop. Понимание всей картины помогает объяснить, почему HDFS проектировался именно так.

Четыре ключевых компонента Hadoop¶

HDFS - уровень хранения данных. Отвечает за: распределённое хранение файлов, репликацию для отказоустойчивости, блочную структуру для параллельного доступа.

MapReduce - исторически первый движок обработки данных над HDFS. Его модель: Map-фаза разбивает входные данные на key-value пары и обрабатывает их параллельно, Reduce-фаза агрегирует промежуточные результаты в финальный ответ. MapReduce пишет промежуточные данные на диск между фазами, что делает его медленнее Spark (который держит данные в памяти).

YARN (Yet Another Resource Negotiator) - менеджер ресурсов кластера. Когда вы запускаете Spark-приложение на Hadoop-кластере через yarn master, именно YARN решает: на каких физических серверах запустить Spark Executors, сколько памяти и CPU им выделить. YARN позволяет запускать несколько приложений одновременно, разделяя ресурсы кластера.

Hadoop Common - общая инфраструктура: Java-утилиты, файловые абстракции (интерфейс FileSystem, который абстрагирует и HDFS, и S3, и локальную FS), нативные библиотеки (libhadoop.so для Short-Circuit Reads).

Экосистема инструментов поверх Hadoop¶

Большинство этих инструментов были созданы для работы с MapReduce поверх HDFS. С появлением Spark часть из них устарела, но HDFS как уровень хранения остаётся актуальным на тысячах enterprise-кластеров.

Компоненты HDFS: разделение обязанностей¶

HDFS построена на классической master-worker архитектуре. Один NameNode (мастер) управляет пространством имён и метаданными всей файловой системы. Множество DataNode (воркеры) физически хранят блоки данных на своих локальных дисках. Такое разделение позволяет масштабировать объём хранения линейно (добавляя DataNode), не изменяя логику управления метаданными.

Схема показывает полную картину архитектуры HDFS с HA. Обратите внимание на ключевой принцип: клиенты получают метаданные от NameNode, но данные читают и пишут напрямую в DataNode. NameNode никогда не является узким местом для data-трафика - он управляет только каталогом.

NameNode: мозг кластера¶

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

Что NameNode хранит в оперативной памяти:

• Filesystem namespace - полное дерево директорий и файлов (аналог inode-таблицы в Linux). Каждая директория, каждый файл, права доступа, владелец, время создания и изменения.
• Block Map - для каждого блока: список DataNode, на которых хранятся его реплики. Эта структура НЕ персистируется - она восстанавливается из Block Reports DataNode при каждом старте.
• DataNode Registry - список живых DataNode и их атрибуты (rack, свободное место, heartbeat-статус, последнее время контакта).

Ключевая проблема: каждый файл и каждый блок стоит ~150 байт в памяти NameNode. Это кажется мелочью, но на реальных Data Lake это означает:

Data Lake с миллиардом мелких файлов потребует 600 GB RAM только для хранения метаданных - это уже за пределами разумного. Именно поэтому компакция файлов в Data Lake (OPTIMIZE в Delta Lake, rewrite_data_files в Iceberg) - это не просто ускорение чтения, но и забота о здоровье NameNode.

Функции NameNode в деталях:

• Управление filesystem операциями: открытие и закрытие файлов, создание и удаление директорий, переименование
• Управление клиентским доступом: контроль прав доступа, блокировки файлов при записи
• Управление репликацией: решение куда разместить реплики (Rack Awareness), запуск под-репликации при потере узлов
• Управление DataNode: обработка heartbeat, блокировка и разблокировка DataNode

FsImage и EditLog: персистентность метаданных¶

NameNode хранит метаданные в памяти для скорости. Но после рестарта эта память пуста - нужно восстановить состояние с диска. Для этого существуют два механизма:

FsImage - это полный снимок (snapshot) файловой системы на конкретный момент времени, сохранённый на диске в бинарном формате. При старте NameNode загружает FsImage в память как начальное состояние. FsImage компактный, загружается быстро.

EditLog - это журнал транзакций (write-ahead log), куда NameNode записывает каждое изменение файловой системы: создание файла, удаление, переименование, аллокация блока. При старте NameNode воспроизводит все записи из EditLog поверх FsImage, чтобы получить актуальное состояние.

Проблема длинного EditLog. Если NameNode работает долго без рестарта, EditLog растёт. При следующем старте ему придётся воспроизвести миллионы операций, и запуск займёт десятки минут. Чтобы этого избежать, существует процедура checkpointing: периодически EditLog и FsImage сливаются в новый FsImage, а EditLog обнуляется. В конфигурации HA (High Availability) эту функцию выполняет Standby NameNode; в старой архитектуре - Secondary NameNode (который, несмотря на название, не является горячим резервом и не может принять управление при падении Primary).

При старте NameNode:

1. Загружает FsImage в RAM - быстро (бинарный снимок)
2. Воспроизводит все записи из EditLog поверх - это займёт тем больше времени, чем длиннее EditLog
3. Принимает Block Reports от DataNode, строит Block Map
4. Только после этого выходит из Safe Mode и начинает обслуживать запросы

DataNode: хранитель блоков¶

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

Физическое хранение. Каждый блок хранится как обычный файл на локальной файловой системе DataNode (ext4, xfs). Рядом с каждым блоком лежит файл метаданных блока (.meta) с контрольными суммами (checksums). При каждом чтении DataNode проверяет checksum - если данные повреждены (disk corruption, bitflip), DataNode сообщает NameNode, что этот блок испорчен, и NameNode инициирует копирование реплики с другого DataNode.

Функции DataNode в деталях:

• Хранение и чтение блоков: физическое хранение блоков данных на локальных дисках
• Реплицирование блоков: создание и удаление реплик по командам NameNode
• Верификация целостности: проверка checksums при каждом чтении
• Heartbeat: периодическое сообщение NameNode о своём состоянии
• Block Report: отчёт о хранящихся блоках при старте и периодически
• Взаимодействие с соседними DataNode: прямая пересылка блоков в pipeline при записи

Организация данных на диске:

/data/hdfs/
├── current/
│   ├── BP-123456789/          # Block Pool ID
│   │   ├── current/
│   │   │   ├── finalized/
│   │   │   │   └── subdir0/
│   │   │   │       ├── blk_1073741825         # блок данных
│   │   │   │       ├── blk_1073741825.meta    # checksums
│   │   │   │       ├── blk_1073741826
│   │   │   │       └── blk_1073741826.meta
│   │   │   └── rbw/           # Replica Being Written
│   │   └── tmp/

Директория rbw (Replica Being Written) - это буфер для блоков, которые сейчас записываются клиентом. Пока запись не подтверждена, блок находится в rbw. После успешной записи перемещается в finalized.

Анатомия хранения данных: блоки и репликация¶

Концепция блоков: единица адресации в HDFS¶

HDFS хранит файлы не целиком, а разбивает на блоки фиксированного размера. Дефолтный размер - 128 MB (в старых версиях Hadoop - 64 MB, в современных production-установках часто 256 MB или 512 MB). Каждый блок хранится независимо и может располагаться на разных DataNode.

Почему 128 MB, а не 4 KB как в обычной файловой системе ОС?

Это принципиальное дизайнерское решение, оптимизированное под batch-аналитику, а не под случайный доступ:

1. Минимизация количества записей в NameNode. Файл 1 GB с блоком 4 KB создаёт 262 144 записей в памяти NameNode. С блоком 128 MB - всего 8 записей. При миллионах файлов разница колоссальная.

2. Sequential I/O доминирует над random I/O. Аналитические джобы читают файлы последовательно от начала до конца. При большом блоке накладные расходы на поиск начала блока (seek()) пренебрежимо малы по сравнению с временем последовательного чтения.

3. Устойчивый TCP-поток. Читая блок 128 MB, клиент устанавливает одно TCP-соединение и получает данные большим непрерывным потоком. При блоке 4 KB пришлось бы устанавливать тысячи соединений - сетевой overhead убил бы пропускную способность.

4. Скорость передачи данных. HDFS поддерживает скорость передачи до 2 GB/s для крупных блоков. При мелких блоках накладные расходы на установку соединений делают эту скорость недостижимой.

Важный нюанс: файл меньше одного блока не занимает полный блок на диске. Файл 5 MB с блоком 128 MB занимает ровно 5 MB места на DataNode. Размер блока - это лишь единица адресации, а не единица выделения места.

Стратегия репликации и Rack Awareness¶

По умолчанию каждый блок хранится в трёх копиях (dfs.replication = 3). Это не просто резервирование - алгоритм размещения реплик (Replica Placement Policy) учитывает физическую топологию сети, чтобы обеспечить как отказоустойчивость, так и оптимальный сетевой трафик при записи.

Стандартное правило размещения трёх реплик:

1. Первая реплика - на том же узле, что пишет клиент (или на случайном узле, если клиент - не DataNode). Это минимизирует сетевой трафик при первоначальной записи.

2. Вторая реплика - на другом rack (серверной стойке), на случайном узле. Это ключевой момент: если упадёт целый rack (авария питания, сетевой коммутатор) - данные выживут, потому что есть реплика в другом rack.

3. Третья реплика - на том же rack, что вторая реплика, но на другом узле. Это снижает стоимость межрэковой репликации (data идёт через внешний коммутатор только один раз) при сохранении rack-level fault tolerance.

Почему такая схема оптимальна? Рассмотрим сценарии отказов:

• Падение одного DataNode → 2 реплики выживают (на DN1 и DN3/DN4). NameNode запускает фоновую репликацию.
• Падение целого Rack 1 → DN3 и DN4 содержат 2 реплики. Данные доступны только из одного rack, но уже безопасны.
• Одновременное падение Rack 1 и одного узла в Rack 2 → одна реплика выжила. Данные доступны.

Эта схема обеспечивает сохранность данных при любом одиночном отказе (узел или rack) и при большинстве двойных отказов.

Рабочий цикл HDFS: как всё функционирует¶

Понимание рабочего цикла HDFS помогает диагностировать проблемы производительности и предсказывать поведение системы под нагрузкой.

Девять ключевых преимуществ HDFS¶

Архитектурные решения HDFS дают конкретные операционные преимущества:

1. Отказоустойчивость (Fault Tolerance). Автоматическое обнаружение аппаратных сбоев через heartbeat и Block Reports. При потере DataNode NameNode немедленно инициирует репликацию недостающих блоков на другие узлы - всё это происходит автоматически без вмешательства оператора.

2. Высокая пропускная способность. HDFS поддерживает скорость передачи до 2 GB/s при последовательном чтении крупных файлов. Это достигается за счёт pipeline-архитектуры: данные текут непрерывным потоком от DataNode к клиенту без промежуточной буферизации на NameNode.

3. Потоковый доступ к данным. HDFS оптимизирован для паттерна «write once, read many times». Аналитические джобы читают файлы последовательно от начала до конца - именно под этот паттерн оптимизирован I/O DataNode.

4. Переносимость. HDFS разработан на Java и работает на любом оборудовании и операционной системе, поддерживающей JVM. Один и тот же код приложения работает на кластере из серверов разных производителей.

5. Горизонтальное и вертикальное масштабирование. Добавить новый DataNode - это добавить новый сервер и запустить на нём процесс DataNode. NameNode автоматически включит его в кластер. Масштабирование линейное: удвоили количество DataNode - удвоили ёмкость и пропускную способность.

6. Локальность данных (Data Locality). HDFS знает о физическом расположении каждого блока на каждом узле и предоставляет эту информацию вычислительным движкам. Spark использует это для запуска задач прямо там, где лежат нужные блоки - данные не гоняются по сети.

7. Экономическая эффективность. Отсутствие лицензионных сборов (open source), возможность использовать commodity-серверы вместо дорогого специализированного железа, виртуализация хранилища снижает накладные расходы на метаданные. По оценкам NSA, Hadoop превзошёл дорогостоящие проприетарные альтернативы при анализе огромных объёмов данных.

8. Большая ёмкость хранения. HDFS способен хранить и обрабатывать произвольно большие объёмы данных - от гигабайт до экзабайт. Реальные production-кластеры в крупных компаниях хранят десятки и сотни петабайт.

9. Гибкость без предобработки. HDFS хранит данные в том формате, в котором они пришли - без предварительной обработки или схематизации. CSV, JSON, Parquet, ORC, Avro, изображения, видео - всё может лежать рядом и всё доступно через один API.

Жизнеобеспечение кластера: Heartbeat, Block Reports и восстановление¶

Heartbeat: пульс кластера¶

Каждые 3 секунды каждый DataNode отправляет NameNode короткое сообщение: «Я живой». Это называется heartbeat. NameNode ведёт таймер для каждого DataNode. Если за 10 минут (дефолт dfs.namenode.heartbeat.recheck-interval = 5 минут + dfs.heartbeat.interval × 10) от DataNode не было ни одного heartbeat - NameNode объявляет этот DataNode мёртвым (Dead).

Heartbeat - это не просто «я жив». В ответ на heartbeat NameNode может послать DataNode команды:

• Реплицировать конкретный блок на другой DataNode
• Удалить излишнюю реплику (over-replication)
• Выполнить recover блока

Block Reports: инвентаризация блоков¶

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

После старта DataNode периодически (каждые 6 часов по умолчанию, dfs.blockreport.intervalMsec) отправляет инкрементальный Block Report - только изменения с последней отправки. Это позволяет NameNode держать актуальную карту всего кластера без постоянного опроса каждого DataNode.

Under-replication и Over-replication¶

Under-replication - ситуация, когда у блока меньше реплик, чем задано dfs.replication. Причины: DataNode умер, диск вышел из строя, узел временно недоступен. NameNode обнаруживает это через Block Report или через потерю heartbeat и ставит задачу на фоновую репликацию: находит DataNode с живой репликой и командует ему скопировать блок на другой DataNode.

Over-replication - лишние реплики. Происходит, если DataNode, считавшийся мёртвым, вернулся в кластер, и блоки уже были реплицированы. NameNode выбирает «лишнюю» реплику и командует её DataNode удалить. При выборе реплики для удаления NameNode предпочитает удалять реплику на том DataNode, который имеет наименьшее свободное место или наибольшую нагрузку.

Safe Mode¶

При запуске NameNode входит в Safe Mode - особый режим только для чтения. В Safe Mode NameNode:

• Принимает Block Reports от DataNode
• Строит Block Map
• Проверяет, что у каждого блока достаточно реплик

NameNode не выходит из Safe Mode, пока процент блоков с достаточной репликацией не достигнет порогового значения (дефолт 99.9%, dfs.namenode.safemode.threshold-pct). Это предотвращает ситуацию, когда клиенты начинают писать новые данные до того, как NameNode убедился в целостности существующих.

При нормальной работе Safe Mode длится 30–60 секунд после старта. Если кластер потерял много DataNode (например, после массовой замены железа), Safe Mode может длиться минуты - NameNode ждёт, когда достаточное количество DataNode отрапортует о своих блоках.

Проверить статус Safe Mode: hdfs dfsadmin -safemode get

Пайплайн записи: как данные попадают в HDFS¶

Понимание write path важно для понимания того, почему запись в HDFS надёжна, даже если один из узлов упадёт в процессе.

Шаги записи файла:

1. Клиент → NameNode: открытие файла для записи. NameNode создаёт запись в namespace и возвращает клиенту уникальный идентификатор блока и список DataNode для pipeline.

2. NameNode выбирает Pipeline. Применяя Rack Awareness, NameNode выбирает 3 DataNode (при replication=3) и возвращает их IP-адреса клиенту в определённом порядке: DN1 (локальный) → DN2 (другой rack) → DN3 (тот же rack, что DN2).

3. Клиент устанавливает pipeline. Клиент соединяется с DN1, DN1 соединяется с DN2, DN2 соединяется с DN3. Это цепочка TCP-соединений, а не звезда.

4. Потоковая запись пакетами. Клиент разбивает блок на пакеты (~64 KB) и отправляет их в DN1. DN1 сразу пересылает каждый пакет DN2, DN2 - DN3. Все три DataNode записывают данные на диск параллельно.

5. Acknowledgment обратной волной. DN3 подтверждает получение пакета → DN2 получает подтверждение от DN3 и отправляет своё → DN1 получает от DN2 и отправляет своё → клиент получает подтверждение от DN1. Только после этого клиент считает пакет доставленным.

6. Завершение блока. После записи последнего пакета клиент сообщает NameNode, что блок записан успешно. NameNode обновляет метаданные.

Что происходит при сбое узла во время записи?

Если один из DataNode в pipeline падает в процессе записи, клиент получает ошибку от соответствующего узла. Протокол восстановления:

1. Pipeline закрывается
2. Частично записанные данные на выживших DataNode фиксируются
3. Клиент запрашивает у NameNode новый DataNode взамен упавшего
4. Формируется новый pipeline с выжившими DataNode + новый
5. Запись продолжается

Клиентское приложение при этом не видит ошибки (всё обрабатывается прозрачно в HDFS-клиенте), а блок в итоге будет реплицирован до нужного replication factor.

Пайплайн чтения: как Spark читает данные с HDFS¶

Понимание read path объясняет, почему Data Locality критична для производительности.

Шаги чтения файла из Spark:

1. Spark Driver → NameNode: получение метаданных блоков. При построении плана выполнения (df.read.parquet("hdfs://...")) Spark Driver вызывает getBlockLocations() - запрашивает у NameNode список блоков файла и адреса DataNode для каждого блока, отсортированные по близости к читающему узлу.

2. Spark планировщик размещает таски с учётом locality. Каждый блок становится одной входной партицией (если блок ≤ maxPartitionBytes). Планировщик пытается разместить таску на том Executor, который запущен на том же DataNode, что хранит этот блок.

3. Executor → DataNode: прямое чтение. Executor открывает TCP-соединение напрямую к DataNode и читает данные. NameNode больше не участвует в data path.

4. Short-Circuit Local Read. Если Executor запущен на том же физическом узле, что и нужный DataNode, и включена опция dfs.client.read.shortcircuit=true, данные передаются через Unix Domain Socket (или напрямую как файл), минуя TCP-стек полностью. Это критически важная оптимизация - throughput вырастает с ~1 Gbps (сеть) до ~500 MB/s (локальный диск).

Уровни Data Locality в Spark¶

Spark Scheduler пытается достичь наилучшего уровня locality, начиная с самого «близкого» и опускаясь ниже, если нужный Executor занят или недоступен:

Параметр spark.locality.wait (дефолт 3 секунды) определяет, сколько Spark ждёт наилучшего уровня locality перед тем, как опуститься на уровень ниже. При слишком долгом ожидании Spark простаивает; при слишком коротком - теряет locality. Для аналитических джобов с большими данными имеет смысл увеличить до 5–10 секунд.

Увидеть locality в Spark UI можно на вкладке Stages → Tasks: в колонке Locality Level каждая таска показывает свой уровень. Если большинство тасок работают на уровне ANY - это сигнал проблемы: либо Executor размещены на узлах без DataNode, либо DataNode перегружены и не успевают отвечать на запросы, либо данные неравномерно распределены по узлам.

High Availability: Active/Standby NameNode¶

В старой архитектуре HDFS NameNode был единственной точкой отказа (SPOF). Падение NameNode делало весь кластер недоступным. В Hadoop 2.x появился режим HA (High Availability): два NameNode работают параллельно - один Active, один Standby.

Как работает HA:

• Active NameNode обслуживает все клиентские запросы и пишет каждое изменение namespace в JournalNodes (кворумный журнал).
• Standby NameNode непрерывно читает изменения из JournalNodes и применяет их к своей копии namespace. Таким образом, Standby всегда актуален.
• DataNode отправляют heartbeats и Block Reports обоим NameNode одновременно.
• ZooKeeper + ZKFC (ZooKeeper Failover Controller) - отдельный процесс на каждом NameNode, который следит за состоянием «своего» NameNode и участвует в выборах лидера. При падении Active NameNode, ZKFC на Standby обнаруживает это через ZooKeeper и инициирует failover: Standby становится Active.

JournalNodes - кворумный журнал. Минимальная конфигурация: 3 JournalNode. Active NameNode записывает изменение в журнал только если получил подтверждение от большинства JournalNode (floor(N/2)+1 = 2 из 3). Это означает, что потеря одного JournalNode не нарушает работу кластера.

Время failover. При правильно настроенном HA с автоматическим failover через ZKFC, переключение занимает 20–60 секунд. В это время все клиентские операции получают ошибки, но HDFS-клиент (и Spark) реализуют retry-логику и автоматически переподключаются к новому Active NameNode после failover.

HDFS Federation: масштабирование namespace¶

Если даже одного NameNode с 256 GB RAM не хватает под объём метаданных, используют HDFS Federation: несколько независимых NameNode, каждый из которых управляет своей частью namespace (block pool). DataNode регистрируются во всех NameNode и хранят блоки из разных block pools.

Например: NameNode 1 управляет /user/, NameNode 2 управляет /warehouse/, NameNode 3 управляет /tmp/. Каждый NameNode независим, имеет свой EditLog и FsImage. Это горизонтальное масштабирование metadata layer.

Проблема мелких файлов¶

Это одна из наиболее критических операционных проблем в production Data Lake на HDFS. Понимание её причин и последствий обязательно для любого Data Engineer.

Почему мелкие файлы образуются¶

Стриминговые пайплайны. Kafka Consumer каждые 5 минут записывает батч - за сутки это 288 файлов по 10–100 MB на каждый топик. За год - тысячи файлов.

Ошибки проектирования партиционирования. Таблица партиционирована по country (200 стран) × date (365 дней) × hour (24 часа) = 1.7 млн партиций. Если в большинстве партиций мало данных - миллионы мелких файлов.

Spark с дефолтным shuffle.partitions=200. Результат записи большинства Spark-джобов - ровно 200 файлов (по одному на партицию shuffle). При мелких данных это 200 файлов по 100 KB - катастрофа для NameNode.

Последствия мелких файлов¶

1. Нагрузка на NameNode RAM. ~150 байт на файл × 100 млн файлов = 15 GB только под namespace метаданные. Плюс блоки (≈ 1 блок на мелкий файл, ещё 150 байт каждый) - итого 30 GB. При 256 GB RAM на NameNode это 12% памяти - возможно. При 1 млрд файлов - 300 GB - NameNode упадёт.

2. Тысячи тасок в Spark вместо сотен. Каждый мелкий файл - отдельная входная партиция (Input Split). 1 000 000 файлов по 1 MB = 1 000 000 тасок. Планирование и завершение каждой таски имеет overhead ~10–50 ms. 1 млн тасок × 50 ms = 50 000 секунд только на scheduling overhead.

3. Медленный ls и getFileStatus. NameNode обрабатывает ls /warehouse/table/ как транзакцию. При миллионе файлов в директории - эта операция занимает секунды и блокирует другие запросы к NameNode.

Решение: компакция файлов¶

# Антипаттерн: 200 файлов по 100 KB
df.write.parquet("hdfs:///warehouse/table/")

# Правильно: принудительная компакция перед записью
(df
    .coalesce(8)  # 10 GB / 128 MB ≈ 80, возьмём с запасом 8
    .write
    .parquet("hdfs:///warehouse/table/"))

# Или через repartition для балансировки (с shuffle):
(df
    .repartition(80)
    .write
    .parquet("hdfs:///warehouse/table/"))

Правило хорошего тона: целевой размер файла - 128 MB–1 GB (один или несколько блоков HDFS). Файлы меньше 10 MB - почти всегда проблема.

Ограничения и особенности архитектуры HDFS¶

Понимание ограничений HDFS не менее важно, чем понимание его преимуществ. Архитектурные решения, которые сделали HDFS мощным для batch-аналитики, одновременно ограничивают его применимость в других сценариях.

Coupling хранилища и вычислений¶

В HDFS хранение данных тесно связано с вычислительными ресурсами: добавить дисковое пространство означает добавить новые полноценные серверы с CPU и RAM. Это ключевое отличие от объектных хранилищ (S3, MinIO), где хранилище и вычисления масштабируются независимо.

На практике это означает: если Data Lake быстро растёт по объёму, но нагрузка на обработку остаётся прежней - вы всё равно вынуждены покупать новые серверы с CPU и RAM только ради дополнительных дисков. В облачных архитектурах (Spark + S3) этой проблемы нет.

Требование к ребалансировке при добавлении узлов¶

При добавлении новых DataNode HDFS не перемещает существующие данные автоматически. Новые узлы постепенно заполняются по мере записи новых данных, но старые данные остаются на старых узлах. Это создаёт дисбаланс: новые узлы почти пусты, старые заполнены. Для устранения нужно явно запускать hdfs balancer.

HDFS оптимизирован под локально присоединённые диски¶

HDFS предполагает что DataNode имеет прямой доступ к локальным дискам (JBOD, NVMe, SATA SSD). Использование сетевых файловых систем (NFS) или SAN в качестве дискового бэкенда DataNode технически возможно, но теряет большую часть преимуществ локального I/O.

Write-once, read-many семантика¶

HDFS не поддерживает произвольную перезапись блоков (random write) или обновление отдельных записей. Файл можно создать и читать, но изменить уже существующий файл нельзя. Для ACID-операций нужны надстройки: Apache Iceberg, Delta Lake или Apache Hudi.

HDFS vs Object Storage (S3, MinIO, GCS)¶

Современные архитектуры всё чаще выбирают Object Storage вместо HDFS. Понимание разницы - обязательная компетенция Senior Data Engineer.

Критичный момент - Rename при записи Spark. Spark при записи результата сначала пишет во временную директорию /_temporary/, а затем делает атомарный rename финальных файлов. В HDFS rename - O(1) операция на уровне метаданных NameNode. В S3 - rename эмулируется как copy + delete, что при тысяче файлов занимает минуты и не является атомарным. Именно поэтому для S3 нужны специальные механизмы: s3a:// с FileOutputCommitter v2, или Delta Lake / Iceberg с собственными commit-протоколами.

Для on-premise кластеров с требованием data locality (когда compute и storage на одних серверах) HDFS остаётся оптимальным выбором. Для cloud-native архитектуры и Data Lakehouse с Iceberg/Delta, где compute и storage разделены - Object Storage предпочтительнее.

Варианты развёртывания HDFS кластеров¶

В production HDFS кластеры развёртываются тремя принципиально разными способами, каждый из которых имеет своё место.

Single-Node кластер (разработка и тестирование)¶

Все демоны HDFS (NameNode, DataNode, JournalNode) и YARN (ResourceManager, NodeManager) запускаются на одной машине. Используется только для разработки, локального тестирования кода и учёбы. В production не применяется: нет репликации, нет отказоустойчивости.

Multi-Node кластер (production)¶

Стандартная production конфигурация: выделенные серверы под NameNode и JournalNode, отдельные серверы-DataNode для хранения. Минимальная конфигурация HA: 2 NameNode + 3 JournalNode + 3 DataNode. Реальные кластеры насчитывают сотни и тысячи DataNode.

Cloud-based Hadoop¶

Крупные облачные провайдеры предлагают управляемые Hadoop-кластеры: Amazon EMR, Google Dataproc, Azure HDInsight. Они работают поверх облачного объектного хранилища (S3, GCS, ADLS) вместо HDFS на локальных дисках. Это снижает операционные расходы, но устраняет Data Locality.

Реальные применения HDFS в Enterprise¶

HDFS используется в крупных компаниях для решения реальных бизнес-задач:

Ритейл (Marks & Spencer): предиктивная аналитика и управление запасами. Анализ данных о продажах, поведении покупателей и цепочке поставок для оптимизации складских запасов в тысячах магазинов.

Финансы (JPMorgan Chase): моделирование рисков и аналитика клиентских портфелей. Обработка миллионов транзакций в день для выявления мошеннических операций и расчёта кредитных рисков.

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

Государственный сектор (NSA, США): анализ данных разведки. По оценкам NSA, Hadoop превзошёл дорогостоящие проприетарные альтернативы при обработке огромных объёмов данных.

HDFS CLI: практические команды¶

Основные команды работы с файлами¶

# Список файлов и директорий
hdfs dfs -ls /warehouse/table/
hdfs dfs -ls -h /warehouse/table/        # human-readable размеры
hdfs dfs -ls -R /warehouse/             # рекурсивно

# Занятое место
hdfs dfs -du -h /warehouse/table/       # размер на диске (с репликами)
hdfs dfs -du -s -h /warehouse/table/    # суммарный размер

# Копирование
hdfs dfs -put localfile.parquet /warehouse/
hdfs dfs -get /warehouse/file.parquet ./
hdfs dfs -cp /warehouse/src/ /warehouse/dst/

# Удаление
hdfs dfs -rm /warehouse/table/part-001.parquet
hdfs dfs -rm -r /warehouse/old_table/   # рекурсивно

Диагностика и здоровье кластера¶

# Отчёт о состоянии кластера
hdfs dfsadmin -report

# Пример вывода:
# Configured Capacity: 50 TB
# Present Capacity: 47 TB
# DFS Remaining: 32 TB (68.09%)
# DFS Used: 15 TB (31.91%)
# Under replicated blocks: 0
# Blocks with corrupt replicas: 0
# Missing blocks: 0

# Live DataNodes: 12
# Dead DataNodes: 0

# Проверка файловой системы (FSCK)
hdfs fsck /warehouse/table/             # проверить конкретную директорию
hdfs fsck /warehouse/table/ -files -blocks -locations
# Покажет каждый файл, его блоки и на каких DataNode они лежат

# Safe mode
hdfs dfsadmin -safemode get
hdfs dfsadmin -safemode leave  # принудительно выйти из Safe Mode

# Баланс блоков между DataNode
hdfs balancer -threshold 10    # выравнивание пока разница > 10%

Анализ блоков конкретного файла¶

# Посмотреть где лежат блоки файла
hdfs fsck /warehouse/table/file.parquet -files -blocks -locations -racks

# Вывод:
# /warehouse/table/file.parquet 134217728 bytes, replicated: replication=3,
#   1 block(s):  OK
#   0. BP-xxx:blk_1073741825_1001 len=134217728 Live_repl=3
#      [DatanodeInfoWithStorage[192.168.1.10:9866,/default-rack]]
#      [DatanodeInfoWithStorage[192.168.1.20:9866,/rack2]]
#      [DatanodeInfoWithStorage[192.168.1.21:9866,/rack2]]

Связь HDFS с настройками Spark¶

Архитектура HDFS напрямую определяет, как нужно настраивать Spark. Это не абстрактная теория - это конкретные конфиги:

spark.sql.files.maxPartitionBytes и размер блока HDFS¶

# HDFS block size = 128 MB
# spark.sql.files.maxPartitionBytes = 128 MB (128 * 1024 * 1024 = 134217728 байт)
# При чтении файла 1 GB → 8 входных партиций → 8 тасок

spark = SparkSession.builder \
    .config("spark.sql.files.maxPartitionBytes", "134217728") \
    .getOrCreate()

# Проверить фактический параллелизм:
df = spark.read.parquet("hdfs:///warehouse/table/")
print(df.rdd.getNumPartitions())  # должно быть ≈ total_size / 128MB

Когда maxPartitionBytes = HDFS block size, Spark создаёт ровно по одной таске на один блок HDFS. Это оптимально: locality максимальна (одна таска = один блок = один DataNode), а количество тасок предсказуемо.

spark.locality.wait и реакция на занятые DataNode¶

# Дать Spark больше времени ждать NODE_LOCAL locality
# Полезно при высокой нагрузке на DataNode
spark = SparkSession.builder \
    .config("spark.locality.wait", "10s") \
    .config("spark.locality.wait.node", "10s") \
    .config("spark.locality.wait.rack", "5s") \
    .getOrCreate()

Мониторинг HDFS¶

NameNode Web UI¶

NameNode предоставляет web UI на порту 9870 (HDFS 3.x) или 50070 (HDFS 2.x):

http://namenode-host:9870/

Ключевые метрики:

• Cluster Summary: Total capacity, Used, Remaining, Under replicated blocks, Corrupt blocks
• Datanodes: список DataNode, их статус, свободное место, блоки
• Startup Progress: прогресс загрузки при старте NameNode

Ключевые метрики через JMX¶

# NameNode JMX metrics (порт 9870)
curl http://namenode:9870/jmx?qry=Hadoop:service=NameNode,name=NameNodeInfo \
  | python3 -m json.tool | grep -E "CapacityUsed|NumLiveDataNodes|NumberOfMissingBlocks"

# Важные метрики:
# - NumberOfMissingBlocks: 0 - хорошо, > 0 - критично
# - NumberOfUnderReplicatedBlocks: норма небольшое число при постоянных изменениях
# - CapacityUsedGB / CapacityTotalGB: заполненность кластера (критично при > 80%)
# - NumLiveDataNodes: сколько DataNode живых
# - NumDeadDataNodes: сколько мёртвых

Антипаттерны использования HDFS¶

1. Миллионы мелких файлов. Каждый файл < 10 MB в Data Lake - потенциальная проблема. Решение: компакция через coalesce() / repartition() при записи, или периодический OPTIMIZE (Delta Lake) / rewrite_data_files (Iceberg).

2. Очень большие файлы без сплиттинга. Один файл 100 GB - одна таска, которая читает 100 GB последовательно. Весь параллелизм теряется. Оптимум: файлы 128 MB–1 GB, splittable форматы (Parquet, ORC, Avro с sync markers).

3. Случайная запись и перезапись отдельных блоков. HDFS оптимизирован под write-once-read-many семантику. Случайная перезапись блоков (как в транзакционной БД) не поддерживается. Нужна Delta Lake / Iceberg для ACID.

4. Высокий replication factor для холодных данных. replication=3 для архивных данных, к которым обращаются раз в квартал, - излишняя трата дискового пространства (×3 overhead). Используйте Erasure Coding для cold data (урок 05).

5. Игнорирование Rack Awareness. Если HDFS не знает о rack-топологии (не настроен net.topology.script.file.name), все DataNode будут считаться в одном rack (/default-rack). При отказе физического rack потеряете все 3 реплики для части данных.

6. Запись временных данных в HDFS вместо /tmp локальной FS. Spark создаёт временные файлы при shuffle (данные между стадиями). Эти файлы НЕ должны идти в HDFS - они пишутся на локальные диски DataNode/Executor через spark.local.dir.

Практика¶

Задание 1: исследование состояния кластера¶

Используя CLI, выполните диагностику учебного кластера:

# 1. Общий отчёт по кластеру
hdfs dfsadmin -report 2>&1 | head -30

# 2. Проверить файловую систему
hdfs fsck /user -files -blocks 2>&1 | tail -20

# 3. Найти мелкие файлы (< 10 MB)
hdfs dfs -ls -R /warehouse/ | awk '{print $5, $8}' | sort -n | head -20
# Колонка 5 - размер в байтах, колонка 8 - путь
# Файлы с размером < 10485760 (10 MB) - кандидаты на компакцию

# 4. Посмотреть использование места по директориям
hdfs dfs -du -h -s /warehouse/*

Задание 2: анализ Data Locality в Spark UI¶

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HDFSLocalityDemo") \
    .config("spark.locality.wait", "5s") \
    .getOrCreate()

spark.sparkContext.setLogLevel("INFO")

df = spark.read.parquet("hdfs:///warehouse/table/")
print(f"Начальных партиций: {df.rdd.getNumPartitions()}")

result = df.select("col1", "col2").where("col3 > 100").count()
print(f"Результат: {result}")

# В Spark UI (порт 4040):
# Stages → Stage 0 → Tasks
# Посмотреть колонку "Locality Level"
# Если большинство тасок RACK_LOCAL или ANY → проблема locality

Задание 3: сравнение производительности мелких vs крупных файлов¶

import time
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("SmallFilesDemo").getOrCreate()

df = spark.range(0, 10_000_000).select(
    F.col("id"),
    F.rand().alias("value"),
    (F.col("id") % 100).alias("category")
)

# Сценарий 1: 200 мелких файлов (дефолт)
start = time.time()
df.write.mode("overwrite").parquet("hdfs:///tmp/small_files_test/")
write_time_small = time.time() - start
print(f"Запись 200 файлов: {write_time_small:.1f}s")

# Сценарий 2: 8 крупных файлов
start = time.time()
df.repartition(8).write.mode("overwrite").parquet("hdfs:///tmp/large_files_test/")
write_time_large = time.time() - start
print(f"Запись 8 файлов: {write_time_large:.1f}s")

# Сравнение чтения
start = time.time()
spark.read.parquet("hdfs:///tmp/small_files_test/").count()
read_small = time.time() - start

start = time.time()
spark.read.parquet("hdfs:///tmp/large_files_test/").count()
read_large = time.time() - start

print(f"\nЧтение 200 файлов: {read_small:.1f}s")
print(f"Чтение 8 файлов:   {read_large:.1f}s")
print(f"Ускорение: {read_small/read_large:.1f}x")

Чек-лист и связь с настройками Spark¶

Итоговый чек-лист¶

• HDFS возник из Google File System и Apache Nutch, создан Doug Cutting и Mike Cafarella в 2002 году - это решение реальной инженерной задачи, а не академическая разработка
• HDFS - только один компонент экосистемы Hadoop: рядом работают YARN (ресурсы), MapReduce/Spark (вычисления), ZooKeeper (координация)
• HDFS разделяет метаданные (NameNode) и данные (DataNode) - NameNode никогда не является узким местом для data-трафика
• Каждый файл и блок стоит ~150 байт RAM на NameNode - миллионы мелких файлов = OOM NameNode
• Размер блока 128 MB оптимален для батч-аналитики: минимизирует метаданные, максимизирует sequential I/O
• Rack Awareness защищает от потери rack: реплики 1 (local) + 2 (remote rack) + 3 (remote rack, другой узел)
• Spark читает блоки напрямую с DataNode, NameNode участвует только в обмене метаданными
• Data Locality: PROCESS_LOCAL > NODE_LOCAL > RACK_LOCAL > ANY - цена каждого уровня ниже на порядок
• spark.sql.files.maxPartitionBytes = HDFS block size → одна таска = один блок = максимальная locality
• HA с Active/Standby NameNode + JournalNodes устраняет SPOF при правильной настройке ZooKeeper
• HDFS не поддерживает произвольную перезапись - для ACID-операций нужны Delta Lake, Iceberg или Hudi
• Для cold data - Erasure Coding вместо репликации × 3 (экономит 50% дискового места)
• Short-Circuit Local Reads - обязательная оптимизация: данные через Unix socket вместо TCP, ~5× быстрее
• Coupling compute+storage - главный аргумент в пользу S3/MinIO для современных Cloud-Native архитектур