HDFS CLI: диагностика через hdfs dfs, fsck, dfsadmin и балансировка

Полное руководство по HDFS CLI для Data Engineer: навигация и аудит через hdfs dfs, диагностика целостности через fsck, администрирование кластера через dfsadmin, квоты, SafeMode, балансировка HDFS Balancer, влияние дисбаланса на Spark и сценарии диагностики реальных аварий.

storage platform

1. Архитектурный контекст: зачем Data Engineer знать HDFS CLI¶

Spark - это слой вычислений. HDFS - это слой хранения. В идеальном мире между ними нет трения: Spark читает данные, выполняет трансформации, пишет результаты. Но production-кластеры не живут в идеальном мире. Диски заканчиваются, DataNode падают, сетевые партиции вызывают потерю реплик, квоты заполняются - и всё это немедленно отражается в логах Spark в виде загадочных исключений.

Специалист, который умеет работать только на уровне DataFrame API, при первом же инфраструктурном инциденте окажется беспомощным. Умение быстро спуститься на уровень HDFS CLI и поставить диагноз - это навык, который отличает Senior Data Engineer от Junior.

Экосистема утилит: кто за что отвечает¶

Как распознать инфраструктурную аварию по логам Spark¶

Прежде чем идти в CLI, нужно понять что именно сломалось. Spark-логи дают подсказки:

Исключение в логах Spark	Вероятная причина	Диагностика
`BlockMissingException`	Блок недоступен (DataNode упал или диск сломан)	`hdfs fsck /path -blocks -locations`
`IOException: All replicas are corrupt`	Все реплики блока повреждены (bit rot, аппаратный сбой)	`hdfs fsck /path -files -blocks`
`RemoteException: org.apache.hadoop.hdfs.server.namenode.SafeModeException`	NameNode в SafeMode	`hdfs dfsadmin -safemode get`
`AccessControlException`	Неверные права доступа	`hdfs dfs -ls /path` → проверить права
`QuotaExceededException`	Превышена квота на файлы или пространство	`hdfs dfs -count -q /path`
`StandbyException`	Spark подключился к Standby NameNode	`hdfs haadmin -getServiceState nn1`
Медленное чтение без ошибок	Дисбаланс DataNode или Small Files	`hdfs dfsadmin -report` + `hdfs fsck`

2. Пользовательский слой: глубокий аудит данных через hdfs dfs¶

hdfs dfs - это командный интерфейс для работы с HDFS файловой системой. Синтаксически он почти идентичен Linux shell-командам, поэтому обучение происходит быстро. Но семантика имеет важные отличия.

Навигация и исследование: базовый синтаксис¶

# ── Просмотр содержимого директории ─────────────────────────────────
# -l (long listing): показывает права, владельца, размер, дату
# -h (human-readable): размеры в KB/MB/GB вместо байт
# -R (recursive): рекурсивный обход поддиректорий

hdfs dfs -ls /user/spark/data/
# drwxr-xr-x  - spark hadoop 0 2024-01-15 10:00 /user/spark/data/bronze
# drwxr-xr-x  - spark hadoop 0 2024-01-14 22:00 /user/spark/data/silver
# drwxr-xr-x  - spark hadoop 0 2024-01-14 22:30 /user/spark/data/gold

hdfs dfs -ls -h /user/spark/data/bronze/events/
# Первая колонка:  права (drwxrwxr-x)
# Вторая:          число реплик (3 для файлов, - для директорий)
# Третья:          владелец
# Четвёртая:       группа
# Пятая:           размер (human-readable с -h)
# Шестая-Седьмая:  дата/время последнего изменения
# Восьмая:         путь

# Рекурсивный просмотр (осторожно на больших Data Lake!)
hdfs dfs -ls -R /user/spark/data/bronze/ | head -20

# ── Поиск файлов ──────────────────────────────────────────────────────
# -find аналог linux find с поддержкой шаблонов и условий по времени

# Все Parquet файлы в директории
hdfs dfs -find /user/spark/data/ -name "*.parquet"

# Файлы изменённые за последние 24 часа
hdfs dfs -find /user/spark/data/ -mmin -1440

# Файлы старше 30 дней (для поиска кандидатов на архивацию)
hdfs dfs -find /user/spark/data/bronze/ -mtime +30 -type f

# Просмотр содержимого текстовых файлов (CSV, JSON, логи)
hdfs dfs -cat /user/spark/data/raw/events_001.json | head -5
hdfs dfs -tail /user/spark/data/logs/processing.log

# Чтение сжатых и бинарных форматов
hdfs dfs -text /user/spark/data/raw/events.avro | head -5
# -text понимает: gzip, bzip2, lz4, Avro, SequenceFile

Анализ распределения дискового пространства¶

Понимание реального объёма данных - ключевая задача Data Engineer. hdfs dfs -du и hdfs dfs -df дают разные срезы информации:

# ── hdfs dfs -du: размер директорий и файлов ─────────────────────────
# Показывает два числа: DISK_SPACE (с репликами) и DATA_SIZE (без реплик)

hdfs dfs -du -h /user/spark/data/
# 125.3 G  376.0 G  /user/spark/data/bronze   ← 125 GB данных, 376 GB с 3x репликами
# 89.5 G   268.5 G  /user/spark/data/silver
# 12.1 G   36.3 G   /user/spark/data/gold

# -s (summary): суммарный размер всей директории
hdfs dfs -du -s -h /user/spark/data/
# 226.9 G  680.8 G  /user/spark/data

# Без -h - в байтах (для скриптов и мониторинга)
hdfs dfs -du -s /user/spark/data/bronze/events/
# 134217728  402653184  /user/spark/data/bronze/events/
#              ^
#              DiskSpaceConsumed (с репликами) - реальный расход дисков DataNode

# ── hdfs dfs -df: свободное место на HDFS ───────────────────────────
# Показывает общую ёмкость кластера, использованное и свободное место
hdfs dfs -df -h /
# Filesystem             Size   Used  Available  Use%
# hdfs://mycluster  120.0 T  68.5 T     51.5 T   57%

# Разница между du и df:
# du показывает сколько займут конкретные файлы/директории
# df показывает общую ёмкость всей файловой системы

Поиск аномалий через hdfs dfs -count¶

hdfs dfs -count - это главная команда для быстрой диагностики Small Files. Она возвращает три числа: DIR_COUNT (директории), FILE_COUNT (файлы), CONTENT_SIZE (суммарный объём данных).

# Базовое использование
hdfs dfs -count /user/spark/data/bronze/events/
# DIR_COUNT  FILE_COUNT  CONTENT_SIZE  PATH
# 366        2856        134217728     /user/spark/data/bronze/events/

# Интерпретация:
# 366 директорий = 365 date-партиций + 1 корневая
# 2856 файлов = ~7.8 файлов на партицию (разумно при 8 Executor'ах)
# 134 MB суммарно = ~46 KB на файл (КРИТИЧНО! Small Files!)

# ── -q (quota): показывает установленные квоты ──────────────────────
hdfs dfs -count -q /user/spark/
# Формат: QUOTA  REMAINING_QUOTA  SPACE_QUOTA  REMAINING_SPACE_QUOTA  DIRS  FILES  SIZE  PATH
# none   inf    none   inf    366  2856   134217728  /user/spark/data/bronze

# none = квота не установлена (inf = неограничено)

# ── -v (verbose): показывает заголовки ─────────────────────────────
hdfs dfs -count -q -v /user/spark/
# QUOTA  REMAINING_QUOTA  SPACE_QUOTA  REMAINING_SPACE_QUOTA  DIR_COUNT  FILE_COUNT  CONTENT_SIZE  PATHNAME

# Скрипт для поиска "больных" директорий (много файлов, мало данных)
hdfs dfs -count -q /user/spark/data/bronze/ | awk '{
    dir_count = $5
    file_count = $6
    size_bytes = $7
    path = $8

    if (file_count > 0) {
        avg_file_size_kb = size_bytes / file_count / 1024
        if (avg_file_size_kb < 1024) {  # < 1 MB на файл
            printf "SMALL FILES: %-60s files=%d avg=%.1f KB\n",
                path, file_count, avg_file_size_kb
        }
    }
}' | sort -k3 -n | head -20

Управление данными: cp, mv, rm, put, get¶

# ── Копирование внутри HDFS ──────────────────────────────────────────
# Важно: -cp выполняется НА стороне кластера, данные не проходят через клиент
# Это COPY, не Move - данные дублируются

hdfs dfs -cp /data/bronze/2024/ /data/archive/2024/

# -f (force): перезаписывать существующие файлы
hdfs dfs -cp -f /tmp/staging/part-001.parquet /data/production/

# -p (preserve): сохранять metadata (timestamps, permissions, replication)
hdfs dfs -cp -p /data/source/ /data/backup/

# ── Перемещение (rename/move) ────────────────────────────────────────
# mv - это атомарная операция переименования в пределах одной FS
# (быстро, нет фактического копирования данных)

hdfs dfs -mv /data/incoming/batch_001/ /data/processed/batch_001/

# ── Удаление ────────────────────────────────────────────────────────
# ВНИМАНИЕ: удалённые файлы попадают в /user/<username>/.Trash/
# Реальное удаление происходит через checkpoint (каждые 24 часа по умолчанию)

hdfs dfs -rm /tmp/staging/old_file.parquet
# Вывод: Moved: '/tmp/staging/old_file.parquet' to trash at:
#   hdfs://cluster/user/spark/.Trash/Current

# -skipTrash: удалить без корзины (НЕОБРАТИМО!)
hdfs dfs -rm -skipTrash /tmp/staging/old_file.parquet

# -r (recursive): удалить директорию рекурсивно
hdfs dfs -rm -r /tmp/staging/

# Очистить корзину вручную
hdfs dfs -expunge

# ── Загрузка и скачивание ────────────────────────────────────────────
# put: копирование с локальной ФС в HDFS
hdfs dfs -put /local/data/events.csv /user/spark/data/raw/
hdfs dfs -put -f /local/data/events.csv /user/spark/data/raw/  # overwrite

# get: копирование из HDFS на локальную ФС
hdfs dfs -get /user/spark/data/raw/events.csv /local/download/
# Или целую директорию:
hdfs dfs -get /user/spark/data/gold/ /local/export/

# getmerge: скачать все файлы директории в один локальный файл
# Полезно для экспорта небольших аналитических результатов
hdfs dfs -getmerge /user/spark/results/ /local/final_report.csv

Управление правами: POSIX и ACL¶

# ── POSIX права (chmod, chown) ───────────────────────────────────────
# HDFS поддерживает стандартные POSIX права rwxrwxrwx

# Изменение прав доступа
hdfs dfs -chmod 755 /user/spark/data/
hdfs dfs -chmod -R 750 /user/spark/data/bronze/  # рекурсивно

# Числовая нотация и символьная нотация
hdfs dfs -chmod go-w /user/spark/data/gold/      # убрать write для group и other
hdfs dfs -chmod u+x /user/spark/scripts/run.sh   # добавить execute для owner

# Изменение владельца и группы
hdfs dfs -chown spark:hadoop /user/spark/data/
hdfs dfs -chown -R spark:hadoop /user/spark/

# ── ACL (Access Control Lists): тонкое управление ─────────────────────
# POSIX права позволяют задать права только для owner, group, other.
# ACL позволяют задать права для произвольных пользователей и групп.

# Просмотр ACL
hdfs dfs -getfacl /user/spark/data/gold/
# # file: /user/spark/data/gold/
# # owner: spark
# # group: hadoop
# user::rwx
# group::r-x
# other::---
# group:analysts:r-x   ← дополнительный ACL для группы analysts

# Добавление ACL
# Дать read+execute группе analysts на директорию gold:
hdfs dfs -setfacl -m group:analysts:r-x /user/spark/data/gold/

# Дать read+execute пользователю john на конкретный файл:
hdfs dfs -setfacl -m user:john:r-- /user/spark/data/gold/report.parquet

# ACL по умолчанию (применяются к новым файлам в директории)
hdfs dfs -setfacl -m default:group:analysts:r-x /user/spark/data/gold/

# Удаление ACL
hdfs dfs -setfacl -x group:analysts /user/spark/data/gold/

# ВАЖНО: для работы ACL нужно включить в hdfs-site.xml:
# <property>
#   <name>dfs.namenode.acls.enabled</name>
#   <value>true</value>
# </property>

3. Инспекция здоровья метаданных: магия утилиты hdfs fsck¶

hdfs fsck (File System Check) - это инструмент, который проверяет консистентность namespace HDFS без чтения реальных данных с дисков DataNode. Он работает только с метаданными NameNode: проверяет что каждый блок имеет нужное число реплик и что все блоки физически доступны.

Что fsck проверяет и как это работает¶

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

Классификация проблемных блоков¶

Under-replicated blocks - блоки с числом доступных реплик меньше нужного (обычно < 3). Возникают когда:

DataNode упал или недоступен (временно или постоянно)
Диск DataNode сломан
DataNode добавлен в кластер и данные ещё не сбалансированы на него

Under-replicated - не катастрофа (данные ещё читаются), но снижает отказоустойчивость. HDFS автоматически реплицирует такие блоки когда DataNode восстанавливается.

Over-replicated blocks - блоки с числом реплик больше нужного. Возникают когда DataNode возвращается после падения, и его блоки «дублируют» реплики, созданные HDFS для восстановления. HDFS автоматически удаляет лишние реплики.

Corrupt blocks - блоки где checksum данных не совпадает с ожидаемым. Признак повреждения данных на диске (bit rot, аппаратный сбой). Это серьёзная проблема - читать такой блок нельзя.

Missing blocks - блоки которых нет ни на одном доступном DataNode. Если replication factor = 3 и все три сервера с репликами упали - блок «пропадает». Это потеря данных - восстановить их невозможно без резервной копии.

Полный набор fsck команд¶

# ── Базовая проверка ─────────────────────────────────────────────────
# Сканирует всё пространство имён, выводит статистику
hdfs fsck /
# Вывод:
# Connecting to namenode via http://namenode:9870/fsck?...
# FSCK started by spark on Mon Jan 15 10:00:00 2024...
# ...................
# Status: HEALTHY
# Number of data-nodes:  9
# Number of racks:       3
# Total dirs:            1234
# Total symlinks:        0
# Total files:           847392
# Total blocks (validated): 2153891
# Minimally replicated blocks: 2153891 (100.00 %)
# Over-replicated blocks: 0 (0.00 %)
# Under-replicated blocks: 0 (0.00 %)
# Mis-replicated blocks:  0 (0.00 %)
# Default replication factor:  3
# Average block replication:  3.0000
# Missing blocks:          0
# Corrupt blocks:          0
# Missing replicas (global): 0 (0.00 %)
# The filesystem under path '/' is HEALTHY

# ── Проверка конкретного пути ────────────────────────────────────────
hdfs fsck /user/spark/data/bronze/events/date=2024-01-15/

# ── Детальная трассировка: файлы + блоки + физические расположения ───
# САМАЯ ПОЛЕЗНАЯ КОМАНДА для диагностики DATA LOCALITY проблем
hdfs fsck /user/spark/data/bronze/events/date=2024-01-15/part-001.parquet \
  -files -blocks -locations

# Пример вывода:
# /user/spark/data/bronze/events/date=2024-01-15/part-001.parquet 134217728 bytes,
#   replicated: replication=3, 1 block(s):
# 0. BP-123456789-192.168.1.1-123/blk_001 len=134217728 repl=3
#    [datanode1.example.com:50010,
#     datanode4.example.com:50010,
#     datanode7.example.com:50010]
#
# Status: HEALTHY

# ── Вывод только проблемных файлов ──────────────────────────────────
# -openforwrite: показать файлы открытые для записи (незавершённые upload)
# -listCorruptFileBlocks: только файлы с corrupt блоками
# Очень полезно в аварийных ситуациях!
hdfs fsck / -listCorruptFileBlocks
# /user/spark/data/bronze/events/date=2024-01-08/part-003.parquet
# /user/spark/data/bronze/events/date=2024-01-08/part-007.parquet
# The filesystem under path '/' has 2 CORRUPT files

# ── Дополнительные ключи ─────────────────────────────────────────────
# -includeSnapshots: включить снапшоты в проверку
# -racks: показать rack-информацию для каждого блока
# -noTopology: не показывать топологию rack'ов (ускоряет вывод)
# -blockId <blk_id>: проверить конкретный блок по его ID

# Полезный скрипт: поиск всех under-replicated файлов
hdfs fsck / 2>&1 | grep -i "under\|corrupt\|missing" | head -20

Трассировка блока до файла Parquet: практическая диагностика¶

Когда Spark упал с BlockMissingException: Could not obtain block BP-123.../blk_001_1234, нужно выяснить: какой файл содержит этот блок?

# Шаг 1: Найти файл по ID блока
# Block ID из ошибки: blk_001_1234
# В новых версиях HDFS:
hdfs fsck / -blockId blk_001_1234
# Вывод:
# Block blk_001_1234 belongs to:
# /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet

# Шаг 2: Посмотреть все блоки этого файла и их расположение
hdfs fsck /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet \
  -files -blocks -locations -racks
# Вывод покажет: какие DataNode хранят реплики каждого блока
# и в каких rack'ах они находятся

# Шаг 3: Проверить доступность конкретного DataNode
hdfs dfsadmin -report | grep -A 10 "datanode4.example.com"

# Шаг 4: Если DataNode жив но блок Corrupt:
# Попытаться восстановить через удаление corrupt реплик
# и ожидание автоматической репликации с других нод
hdfs dfs -setrep -w 3 /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet
# -w: ждать завершения репликации
# Это принудит HDFS пересчитать реплики

4. Административный слой: управление и лимиты через dfsadmin¶

hdfs dfsadmin - это административный интерфейс для управления кластером HDFS. В отличие от hdfs dfs (который работает от имени пользователя), большинство команд dfsadmin требуют прав суперпользователя HDFS (обычно hdfs или root).

dfsadmin -report: главная команда диагностики кластера¶

hdfs dfsadmin -report

# Типичный вывод (сокращён):
# Configured Capacity: 109951162777600 (100 TB)
# Present Capacity: 105553116979200 (96 TB)
# DFS Remaining: 37580963840000 (34.2 TB)
# DFS Used: 67972153139200 (61.8 TB)
# DFS Used%: 64.5%
# Replicated Blocks:
#   Under replicated blocks: 0
#   Blocks with corrupt replicas: 0
#   Missing blocks: 0
# -------------------------------------------------
# Live datanodes (9):
#
# Name: 192.168.1.11:50010 (datanode1.example.com)
# Hostname: datanode1.example.com
# Decommission Status : Normal
# Configured Capacity: 12216223694848 (11.12 TB)
# DFS Used: 7551870550016 (6.87 TB)
# DFS Used%: 61.82%
# Configured Cache Capacity: 0 (0 B)
# Cache Used: 0 (0 B)
# Last contact: Mon Jan 15 10:00:05 2024
# Last Block Report: Mon Jan 15 09:58:14 2024
#
# Dead datanodes (1):
# Name: 192.168.1.14:50010 (datanode4.example.com)
# Decommission Status : Normal
# Last contact: Mon Jan 15 09:15:33 2024  ← DataNode упал 45 минут назад!

# Полезный grep для быстрой диагностики:
hdfs dfsadmin -report | grep -E "Live|Dead|Under|Missing|Corrupt|DFS Used%"
# Live datanodes (9):
# Dead datanodes (1):
# Under replicated blocks: 1247
# Blocks with corrupt replicas: 0
# Missing blocks: 0
# DFS Used%: 64.5%

Управление квотами: защита NameNode Heap¶

Квоты - это инструмент Data Governance, позволяющий ограничить ресурсы конкретных директорий. Есть два типа: Name Quota (ограничение числа объектов) и Space Quota (ограничение физического объёма).

# ── Name Quota: ограничение числа файлов и директорий ────────────────
# Критически важно для защиты NameNode Heap!
# Задаёт максимальное число объектов (файлов + директорий) в поддереве

# Установить квоту: не более 10000 файлов/директорий в /user/john/
hdfs dfsadmin -setQuota 10000 /user/john/

# Проверить квоту
hdfs dfs -count -q /user/john/
# QUOTA  REMAINING_QUOTA  SPACE_QUOTA  REMAINING_SPACE_QUOTA  DIRS  FILES  SIZE  PATH
# 10000  7234             none         inf                    532   2234   ...   /user/john/
# (использовано 10000 - 7234 = 2766 объектов)

# Очистить квоту (снять ограничение)
hdfs dfsadmin -clrQuota /user/john/

# ── Space Quota: ограничение физического объёма ───────────────────────
# Ограничивает суммарный объём данных (с учётом репликации!)
# Например, установить 100 GB (× replication factor 3 = 300 GB на дисках)

# 100 GB = 107374182400 байт
hdfs dfsadmin -setSpaceQuota 107374182400 /user/john/
# Или более читаемо через команды умножения:
hdfs dfsadmin -setSpaceQuota $((100 * 1024 * 1024 * 1024)) /user/john/

# Проверить Space Quota
hdfs dfs -count -q /user/john/
# QUOTA   REMAINING   SPACE_QUOTA    REMAINING_SPACE_QUOTA  DIRS  FILES  SIZE   PATH
# 10000   7234        322122547200   218103193600           532   2234   ...    /user/john/
#                     ^^^^^^^^^^^^   ^^^^^^^^^^^^
#                     300 GB (100×3) 202 GB свободно

# Очистить Space Quota
hdfs dfsadmin -clrSpaceQuota /user/john/

# ── Практика: автоматическая установка квот для всех пользователей ───
for user in alice bob carol dave; do
    hdfs dfs -mkdir -p /user/$user
    hdfs dfs -chown $user:hadoop /user/$user

    # Name Quota: 1M объектов (защита от runaway jobs)
    hdfs dfsadmin -setQuota 1000000 /user/$user/

    # Space Quota: 5 TB (× 3 репликации = 15 TB на дисках)
    hdfs dfsadmin -setSpaceQuota $((5 * 1024 * 1024 * 1024 * 1024)) /user/$user/

    echo "Квоты установлены для $user"
done

SafeMode: когда NameNode отказывается работать¶

SafeMode - защитный режим NameNode, в котором файловая система работает только на чтение. NameNode автоматически входит в SafeMode при старте (пока не получит Block Report от достаточного числа DataNode) и выходит из него автоматически.

Проблема возникает когда NameNode «застрял» в SafeMode после аварии или перезапуска.

# Проверить текущий статус SafeMode
hdfs dfsadmin -safemode get
# Safe mode is ON  ← NameNode в SafeMode
# Safe mode is OFF ← нормальная работа

# Войти в SafeMode вручную (перед плановыми работами)
hdfs dfsadmin -safemode enter
# Safe mode is ON

# Выйти из SafeMode вручную (после восстановления)
# ВНИМАНИЕ: использовать только если вы уверены что кластер стабилен!
hdfs dfsadmin -safemode leave
# Safe mode is OFF

# Ждать выхода из SafeMode (для скриптов автоматизации)
hdfs dfsadmin -safemode wait
# (зависает пока SafeMode не выключится автоматически или вручную)

# Диагностика причины SafeMode:
# Если кластер не выходит из SafeMode - проверяем блоки:
hdfs dfsadmin -report | grep "Under replicated"
# Under replicated blocks: 15234  ← много под-реплицированных блоков
# HDFS не выходит из SafeMode пока не достигнут threshold

# Threshold задаётся в hdfs-site.xml:
# dfs.namenode.safemode.threshold-pct: 0.999 (99.9% блоков должны быть достаточно реплицированы)
# Если DataNode постоянно недоступен - threshold не достигается → SafeMode вечно

# Решение: если DataNode не восстановится - decommission его
hdfs dfsadmin -decommission datanode4.example.com

5. Балансировка данных (HDFS Balancer): борьба с дисковым перекосом¶

Дисбаланс данных между DataNode - это медленно нарастающая проблема, которая заметна только когда Spark начинает работать медленнее. Понять природу дисбаланса и правильно запустить балансировщик - важный навык для production-администрирования.

Откуда берётся дисбаланс¶

Диагностика дисбаланса перед запуском балансировщика¶

# Просмотр использования дисков по каждому DataNode
hdfs dfsadmin -report | grep -A 5 "Name:"

# Для лучшей читаемости - grep только нужного
hdfs dfsadmin -report 2>/dev/null | awk '
/^Name:/ { current_node = $2 }
/DFS Used%/ {
    gsub(/%/, "", $3)
    printf "%-40s %s%%\n", current_node, $3
}' | sort -k2 -n

# Пример вывода (кластер в плохом состоянии):
# 192.168.1.11:50010 (datanode1)    87.3%  ← перегружен!
# 192.168.1.12:50010 (datanode2)    85.1%  ← перегружен!
# 192.168.1.13:50010 (datanode3)    83.8%  ← перегружен!
# 192.168.1.21:50010 (datanode4)    12.5%  ← новый сервер!
# 192.168.1.22:50010 (datanode5)    13.2%  ← новый сервер!
# Разброс: 87% - 12% = 75% → СИЛЬНЫЙ дисбаланс

# Порог для запуска балансировщика: если разброс > 10%

Запуск и управление HDFS Balancer¶

# ── Базовый запуск ────────────────────────────────────────────────────
# -threshold <n>: допустимая разница в % использования между DataNode
# По умолчанию: 10%
# Балансировщик останавливается когда разница <= threshold

hdfs balancer -threshold 10
# Начнёт перемещать блоки пока max_used - min_used <= 10%

# ── Ограничение полосы пропускания ───────────────────────────────────
# КРИТИЧНО для production!
# Балансировщик без ограничений может утилизировать всю сеть кластера,
# деградируя производительность Spark-джобов.

# Установить лимит: 100 MB/s = 104857600 байт/с
hdfs dfsadmin -setBalancerBandwidth 104857600

# Проверить текущий лимит
hdfs dfsadmin -getBalancerBandwidth
# Balancer bandwidth is 104857600 bytes per second

# Рекомендуемые лимиты:
# В нерабочее время (ночные работы): 200-500 MB/s
# В рабочее время при параллельных Spark Job'ах: 50-100 MB/s
# Критическое восстановление: без ограничений (но только в окно обслуживания)

# ── Запуск в фоне с логированием ─────────────────────────────────────
nohup hdfs balancer -threshold 10 \
  2>&1 | tee /var/log/hadoop/balancer-$(date +%Y%m%d).log &

# Мониторинг прогресса балансировщика
tail -f /var/log/hadoop/balancer-$(date +%Y%m%d).log

# Типичный вывод:
# 2024-01-15 10:00:00,000 INFO balancer.Balancer: Start balancing...
# 2024-01-15 10:00:10,000 INFO balancer.Balancer:
#   Bytes Already Moved: 0
#   Bytes Left To Move:  2.1 TB
#   Bytes Being Moved:   0
# 2024-01-15 10:05:00,000 INFO balancer.Balancer:
#   Bytes Already Moved: 128 GB
#   Bytes Left To Move:  2.0 TB
#   Time Taken:          300 sec
#   Avg Speed:           436 MB/s

# ── Дополнительные параметры ─────────────────────────────────────────
# -idleiterations <n>: число итераций без движения перед остановкой
# -exclude -f <file>: исключить DataNode из балансировки
# -include -f <file>: балансировать только указанные DataNode
# -source <dn_ip:port>: балансировать только с конкретного DataNode

# Пример: балансировать только новые пустые ноды
echo "192.168.1.21:50010
192.168.1.22:50010" > /tmp/new_nodes.txt

hdfs balancer -threshold 10 -include -f /tmp/new_nodes.txt

6. Влияние разбалансировки и битых блоков на производительность Spark¶

Теория CLI-команд имела бы мало ценности без понимания того, как проблемы HDFS отражаются на работе Spark-пайплайнов.

Деградация Data Locality при дисбалансе¶

Когда 3 из 9 DataNode перегружены (85%+ дисков), а 2 новых почти пусты - блоки постепенно перемещаются балансировщиком. Но пока балансировка идёт, возникает эффект «нестабильной локальности»:

Блоки переносятся с datanode1 на datanode5
Spark кешировал расположение блоков с предыдущего запроса
При следующем запуске планировщик отправляет задачи на datanode1 (по устаревшему кешу)
Данных на datanode1 уже нет - задача деградирует до RACK_LOCAL или ANY

Это проявляется в Spark UI как внезапный рост Scheduler Delay и Fetch Wait Time без очевидных изменений в коде.

Каскадные падения Stage при Missing Blocks¶

Рассмотрим полный сценарий аварии и поведения Spark:

Когда Spark сталкивается с missing block:

Executor'а получает BlockMissingException
Spark автоматически делает retry таски (по умолчанию 4 попытки)
Если все попытки fail - Stage помечается как FAILED
Driver выдаёт ошибку: SparkException: Job 0 cancelled because SparkContext was shut down

Анализ логов при таком сценарии:

# Предупреждение о проблемах с блоком (появляется ДО ошибки)
WARN  org.apache.hadoop.hdfs.DFSClient: DFSInputStream has been closed
  by client while it was trying to connect to datanode4.example.com:50010

# Несколько попыток доступа к блоку
INFO  org.apache.hadoop.hdfs.DFSClient: Could not obtain block blk_001
  from datanode4.example.com:50010, trying next DataNode...
INFO  org.apache.hadoop.hdfs.DFSClient: Found 2 datanode(s) still to retry -
  waiting 0ms before next attempt block blk_001, tried: 1

# Финальная ошибка (все реплики недоступны)
ERROR TaskSetManager: Task 42 in stage 3.0 failed 4 times;
  aborting job
org.apache.hadoop.hdfs.BlockMissingException: Could not obtain block:
  BP-123456789/blk_001_1234 file=/user/spark/data/bronze/.../part-005.parquet
    at org.apache.hadoop.hdfs.DFSInputStream.refetchLocations(DFSInputStream.java:827)
    at org.apache.hadoop.hdfs.DFSInputStream.fetchBlockByteRange(DFSInputStream.java:794)
    ...

Что делать при BlockMissingException¶

# Шаг 1: Определить какой файл повреждён
# Из лога: BP-123456789/blk_001_1234
# Найти файл через fsck:
hdfs fsck / -blockId blk_001_1234
# Block blk_001_1234 belongs to:
# /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet

# Шаг 2: Проверить состояние всех блоков этого файла
hdfs fsck /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet \
  -files -blocks -locations

# Шаг 3: Оценить ущерб - есть ли резервная копия партиции?
hdfs dfs -ls /data/archive/bronze/events/date=2024-01-10/

# Шаг 4a: Если есть backup - восстанавливаем
hdfs dfs -cp /data/archive/bronze/events/date=2024-01-10/part-005.parquet \
            /user/spark/data/bronze/events/date=2024-01-10/

# Шаг 4b: Если нет backup - удаляем corrupted файл и перечитываем из источника
hdfs dfs -rm -skipTrash \
  /user/spark/data/bronze/events/date=2024-01-10/part-005.parquet
# Потом перезапустить ingestion pipeline для этой партиции

# Шаг 4c: Если файл не критичен - добавить в список исключений Spark
spark.conf.set("spark.sql.files.ignoreMissingFiles", "true")
# Это позволит Spark игнорировать missing файлы (принимаем потерю данных)

7. Практика: диагностика «убитого» кластера¶

Полный диагностический сценарий¶

Следующий bash-скрипт имитирует работу инженера, который получил alert «Spark Job падает с BlockMissingException» и начинает диагностику.

#!/bin/bash
# hdfs_cluster_diagnosis.sh
# Полная диагностика состояния HDFS кластера
# Запускать при инцидентах или планово раз в день

echo "=========================================="
echo "HDFS CLUSTER DIAGNOSTICS REPORT"
echo "Дата: $(date)"
echo "=========================================="

echo ""
echo "1. ОБЩЕЕ СОСТОЯНИЕ КЛАСТЕРА"
echo "---"

# Краткий отчёт о кластере
REPORT=$(hdfs dfsadmin -report 2>/dev/null)

LIVE_NODES=$(echo "$REPORT" | grep "Live datanodes" | awk '{print $3}' | tr -d '(')
DEAD_NODES=$(echo "$REPORT" | grep "Dead datanodes" | awk '{print $3}' | tr -d '(')
UNDER_REP=$(echo "$REPORT" | grep "Under replicated" | awk '{print $4}')
MISSING=$(echo "$REPORT" | grep "Missing blocks" | awk '{print $3}')
CORRUPT=$(echo "$REPORT" | grep "Blocks with corrupt" | awk '{print $5}')
DFS_USED=$(echo "$REPORT" | grep "DFS Used%:" | head -1 | awk '{print $3}')

echo "Live DataNodes:         $LIVE_NODES"
echo "Dead DataNodes:         $DEAD_NODES"
echo "Under-replicated blocks: $UNDER_REP"
echo "Missing blocks:         $MISSING"
echo "Corrupt blocks:         $CORRUPT"
echo "Cluster DFS Used:       $DFS_USED"

# Оценка критичности
if [ "$DEAD_NODES" != "0" ]; then
    echo "⚠️  ПРЕДУПРЕЖДЕНИЕ: $DEAD_NODES DataNode недоступны!"
fi

if [ "$MISSING" != "0" ]; then
    echo "🚨 КРИТИЧНО: $MISSING блоков отсутствуют! Возможна потеря данных!"
fi

if [ "$CORRUPT" != "0" ]; then
    echo "🚨 КРИТИЧНО: $CORRUPT блоков повреждены!"
fi

echo ""
echo "2. ИСПОЛЬЗОВАНИЕ ДИСКОВ ПО DATANODE"
echo "---"

# Анализ дисбаланса
echo "$REPORT" | awk '
/^Name:/ { current_node = $2 }
/DFS Used%/ {
    gsub(/%/, "", $3)
    used = $3 + 0
    printf "%-45s %5.1f%%", current_node, used
    if (used > 85) printf "  ⚠️  OVERLOADED!"
    printf "\n"
    total += used
    count++
    if (used > max) max = used
    if (min == 0 || used < min) min = used
}
END {
    if (count > 0) {
        printf "\nСредняя загрузка: %.1f%%\n", total/count
        printf "Разброс: %.1f%% (max %.1f%% - min %.1f%%)\n", max-min, max, min
        if (max - min > 20) printf "⚠️  Дисбаланс > 20%%, рекомендуется запустить hdfs balancer\n"
    }
}'

echo ""
echo "3. АНАЛИЗ SMALL FILES (топ-10 проблемных директорий)"
echo "---"

hdfs dfs -count /user/ 2>/dev/null | awk '{
    if ($6 > 10000) {
        avg_size = ($7 + 0) / ($6 + 1) / 1024 / 1024
        printf "%-60s files=%-8d avg=%.1f MB\n", $8, $6, avg_size
    }
}' | sort -k4 -n | head -10

echo ""
echo "4. ПРОВЕРКА ЦЕЛОСТНОСТИ (выборочный fsck)"
echo "---"

# Быстрая проверка - только поиск corrupt файлов
CORRUPT_FILES=$(hdfs fsck / -listCorruptFileBlocks 2>/dev/null | \
    grep -v "^Connecting\|^FSCK started\|^Status\|^Number" | \
    grep "^/" | head -10)

if [ -z "$CORRUPT_FILES" ]; then
    echo "✅ Corrupt файлов не обнаружено"
else
    echo "🚨 Найдены corrupt файлы:"
    echo "$CORRUPT_FILES"
fi

echo ""
echo "5. SAFEMODE"
echo "---"
SAFEMODE=$(hdfs dfsadmin -safemode get 2>/dev/null)
echo "SafeMode статус: $SAFEMODE"
if echo "$SAFEMODE" | grep -q "ON"; then
    echo "⚠️  NameNode в SafeMode! Spark не может писать данные."
fi

echo ""
echo "=========================================="
echo "РЕКОМЕНДАЦИИ:"
echo "---"
if [ "$DEAD_NODES" != "0" ]; then
    echo "• Проверить состояние dead DataNode: hdfs dfsadmin -report"
    echo "• Если нода не восстановится: hdfs dfsadmin -decommission <hostname>"
fi
if [ $(echo "$DFS_USED" | tr -d '%') -gt 85 ] 2>/dev/null; then
    echo "• Кластер заполнен > 85%! Требуется увеличение ёмкости или очистка"
fi
echo "=========================================="

Python-скрипт: автоматизированный аудит HDFS¶

# hdfs_audit.py
# Программный аудит HDFS через JMX и CLI subprocess
# Удобен для интеграции с Alerting системами (Prometheus, Grafana)

import subprocess
import json
import urllib.request
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class DataNodeMetrics:
    hostname: str
    ip_port: str
    used_pct: float
    capacity_gb: float
    used_gb: float
    free_gb: float
    is_dead: bool = False


@dataclass
class ClusterHealthReport:
    live_datanodes: int
    dead_datanodes: int
    under_replicated_blocks: int
    missing_blocks: int
    corrupt_blocks: int
    dfs_used_pct: float
    datanode_metrics: list[DataNodeMetrics] = field(default_factory=list)

    @property
    def is_healthy(self) -> bool:
        return (
            self.dead_datanodes == 0
            and self.missing_blocks == 0
            and self.corrupt_blocks == 0
            and self.dfs_used_pct < 85.0
        )

    @property
    def imbalance_pct(self) -> float:
        if not self.datanode_metrics:
            return 0.0
        used_pcts = [dn.used_pct for dn in self.datanode_metrics if not dn.is_dead]
        if len(used_pcts) < 2:
            return 0.0
        return max(used_pcts) - min(used_pcts)


def run_hdfs_command(*args) -> tuple[str, str, int]:
    """Запускает hdfs команду и возвращает (stdout, stderr, returncode)."""
    result = subprocess.run(
        ["hdfs"] + list(args),
        capture_output=True,
        text=True,
        timeout=60,
    )
    return result.stdout, result.stderr, result.returncode


def get_cluster_health(namenode_host: str, port: int = 9870) -> ClusterHealthReport:
    """
    Получает полный отчёт о здоровье кластера через JMX HTTP API.
    Не требует прав hdfs-суперпользователя.
    """
    # FSNamesystem для основных метрик
    url = f"http://{namenode_host}:{port}/jmx?qry=Hadoop:service=NameNode,name=FSNamesystem"

    try:
        with urllib.request.urlopen(url, timeout=10) as resp:
            data = json.loads(resp.read())
    except Exception as e:
        raise RuntimeError(f"Не удалось подключиться к NameNode JMX: {e}")

    metrics = {}
    for bean in data.get("beans", []):
        if "FSNamesystem" in bean.get("name", "") and "State" not in bean.get("name", ""):
            metrics = bean
            break

    # NameNodeInfo для DataNode списка
    url2 = f"http://{namenode_host}:{port}/jmx?qry=Hadoop:service=NameNode,name=NameNodeInfo"
    with urllib.request.urlopen(url2, timeout=10) as resp:
        data2 = json.loads(resp.read())

    dn_metrics = []
    for bean in data2.get("beans", []):
        if "NameNodeInfo" not in bean.get("name", ""):
            continue

        live_nodes_json = bean.get("LiveNodes", "{}")
        dead_nodes_json = bean.get("DeadNodes", "{}")

        try:
            live_nodes = json.loads(live_nodes_json)
            for hostname, node_data in live_nodes.items():
                capacity = node_data.get("capacity", 0)
                used = node_data.get("usedSpace", 0)
                dn_metrics.append(DataNodeMetrics(
                    hostname=hostname,
                    ip_port=node_data.get("xferaddr", ""),
                    used_pct=(used / capacity * 100) if capacity > 0 else 0,
                    capacity_gb=capacity / 1024**3,
                    used_gb=used / 1024**3,
                    free_gb=(capacity - used) / 1024**3,
                    is_dead=False,
                ))

            dead_nodes = json.loads(dead_nodes_json)
            for hostname in dead_nodes:
                dn_metrics.append(DataNodeMetrics(
                    hostname=hostname, ip_port="",
                    used_pct=0, capacity_gb=0, used_gb=0, free_gb=0,
                    is_dead=True,
                ))
        except json.JSONDecodeError:
            pass

    return ClusterHealthReport(
        live_datanodes=metrics.get("NumLiveDataNodes", 0),
        dead_datanodes=metrics.get("NumDeadDataNodes", 0),
        under_replicated_blocks=metrics.get("UnderReplicatedBlocks", 0),
        missing_blocks=metrics.get("MissingBlocks", 0),
        corrupt_blocks=metrics.get("CorruptBlocks", 0),
        dfs_used_pct=metrics.get("PercentUsed", 0),
        datanode_metrics=dn_metrics,
    )


def print_health_report(report: ClusterHealthReport) -> None:
    """Форматированный вывод отчёта о здоровье кластера."""
    status = "✅ HEALTHY" if report.is_healthy else "❌ UNHEALTHY"
    print(f"\n{'='*60}")
    print(f"HDFS CLUSTER HEALTH: {status}")
    print(f"{'='*60}")
    print(f"Live DataNodes:          {report.live_datanodes}")
    print(f"Dead DataNodes:          {report.dead_datanodes}"
          + ("  ⚠️" if report.dead_datanodes > 0 else ""))
    print(f"Under-replicated blocks: {report.under_replicated_blocks}"
          + ("  ⚠️" if report.under_replicated_blocks > 0 else ""))
    print(f"Missing blocks:          {report.missing_blocks}"
          + ("  🚨" if report.missing_blocks > 0 else ""))
    print(f"Corrupt blocks:          {report.corrupt_blocks}"
          + ("  🚨" if report.corrupt_blocks > 0 else ""))
    print(f"DFS Used:                {report.dfs_used_pct:.1f}%"
          + ("  ⚠️" if report.dfs_used_pct > 85 else ""))
    print(f"DataNode imbalance:      {report.imbalance_pct:.1f}%"
          + ("  ⚠️" if report.imbalance_pct > 20 else ""))

    if report.datanode_metrics:
        print(f"\nDataNode Distribution:")
        for dn in sorted(report.datanode_metrics, key=lambda x: x.used_pct, reverse=True):
            if dn.is_dead:
                print(f"  {dn.hostname:<35}  DEAD  🔴")
            else:
                bar = "█" * int(dn.used_pct / 5)
                print(f"  {dn.hostname:<35}  {dn.used_pct:5.1f}%  {bar}")


# Пример запуска
if __name__ == "__main__":
    report = get_cluster_health("namenode.example.com")
    print_health_report(report)

    if not report.is_healthy:
        print("\n⚠️  Требуется вмешательство!")
        if report.missing_blocks > 0:
            print("  → hdfs fsck / -listCorruptFileBlocks")
        if report.imbalance_pct > 20:
            print(f"  → hdfs dfsadmin -setBalancerBandwidth 104857600")
            print(f"  → hdfs balancer -threshold 10")
        if report.dead_datanodes > 0:
            print("  → hdfs dfsadmin -report (проверить dead ноды)")

Шпаргалка: 20 ключевых HDFS CLI-команд для Data Engineer¶

# ── НАВИГАЦИЯ И ПРОСМОТР ─────────────────────────────────────────────
hdfs dfs -ls -h /path                  # список файлов с размерами
hdfs dfs -du -s -h /path               # суммарный размер директории
hdfs dfs -count -q -v /path            # кол-во файлов + квоты
hdfs dfs -find /path -name "*.parquet" # поиск файлов по шаблону

# ── РАБОТА С ФАЙЛАМИ ─────────────────────────────────────────────────
hdfs dfs -put /local/file /hdfs/path   # загрузка в HDFS
hdfs dfs -get /hdfs/file /local/       # скачивание из HDFS
hdfs dfs -cp -p /src /dst              # копирование с сохранением meta
hdfs dfs -mv /src /dst                 # перемещение (атомарное)
hdfs dfs -rm -skipTrash /path          # удаление без корзины
hdfs dfs -cat /path                    # вывод содержимого файла

# ── ПРАВА ДОСТУПА ────────────────────────────────────────────────────
hdfs dfs -chmod 750 /path              # изменение прав
hdfs dfs -chown user:group /path       # изменение владельца
hdfs dfs -setfacl -m user:john:r-- /p  # добавить ACL

# ── ДИАГНОСТИКА ФАЙЛОВОЙ СИСТЕМЫ ─────────────────────────────────────
hdfs fsck /path -files -blocks -locations  # блоки + DataNode
hdfs fsck / -listCorruptFileBlocks         # только corrupt файлы

# ── АДМИНИСТРИРОВАНИЕ КЛАСТЕРА ───────────────────────────────────────
hdfs dfsadmin -report                  # состояние кластера
hdfs dfsadmin -safemode get/leave      # SafeMode управление
hdfs dfsadmin -setQuota N /path        # Name Quota (файлы)
hdfs dfsadmin -setSpaceQuota N /path   # Space Quota (байты)

# ── БАЛАНСИРОВКА ─────────────────────────────────────────────────────
hdfs dfsadmin -setBalancerBandwidth 104857600  # лимит 100 MB/s
hdfs balancer -threshold 10                    # запуск балансировки
hdfs dfs -setrep -R 2 /path/archive            # изменить replication factor

Итоги: когда и как использовать HDFS CLI¶

HDFS CLI - это не экзотика. Это ежедневный инструмент Data Engineer на production HDFS-кластере.

hdfs dfs используется для работы с данными: просмотр, копирование, загрузка, удаление, управление правами. Первое что открывает инженер при расследовании проблем.

hdfs fsck - диагностика целостности метаданных. Позволяет найти поврежденные файлы, локализовать missing блоки и понять физическое расположение данных. Запускается при BlockMissingException и IOException: All replicas are corrupt.

hdfs dfsadmin - административное управление: отчёт о кластере, квоты, SafeMode. dfsadmin -report должен быть первой командой при любом инциденте - он даёт полную картину состояния кластера за 2 секунды.

hdfs balancer - профилактика деградации. Запускается планово после добавления нод или при выявлении дисбаланса > 20%. Всегда с ограничением полосы пропускания в рабочее время.

Всё вместе: dfsadmin -report → fsck -listCorruptFileBlocks → dfs -du + dfs -count → при необходимости balancer - это стандартная цепочка диагностики любого HDFS-инцидента.