Resolving HDFS Metadata Congestion
at Petabyte Scale

# /etc/hadoop/conf/hadoop-env.sh — NAMENODE JVM TUNING
# Target: 330M files × 1.5 blocks × ~200B overhead → recommend 256–384 GB

export HADOOP_NAMENODE_OPTS="\
  -Xms256g -Xmx256g \
  -XX:+UseG1GC \
  -XX:G1HeapRegionSize=32m \
  -XX:MaxGCPauseMillis=200 \
  -XX:InitiatingHeapOccupancyPercent=35 \
  -XX:+ParallelRefProcEnabled \
  -XX:ConcGCThreads=8 \
  -XX:ParallelGCThreads=16 \
  -XX:+ExplicitGCInvokesConcurrent \
  -XX:+HeapDumpOnOutOfMemoryError \
  -XX:HeapDumpPath=/var/log/hadoop/nn-heapdump.hprof \
  -XX:+PrintGCDetails -XX:+PrintGCDateStamps \
  -Xloggc:/var/log/hadoop/nn-gc.log \
  -Dhadoop.security.logger=INFO,RFAS"

# Increase handler threads to reduce RPC queue depth
# Add to hdfs-site.xml:
# dfs.namenode.handler.count = 3× sqrt(number_of_datanodes)
# For 100 DataNodes → handler.count = 30 (minimum)
# For 500 DataNodes → handler.count = 70

# Step 1: Identify cold partitions (not accessed in 90+ days)
hdfs dfs -stat "%n %Y" /data/logs/2023/* | \
  awk -v cutoff=$(date -d '-90 days' +%s000) '$2 < cutoff {print $1}'

# Step 2: Create HAR archive (runs a MR job)
hadoop archive \
  -archiveName logs_2023_q1.har \
  -p /data/logs/2023 q1 \
  /archives/logs/2023/

# Step 3: Verify archive is readable
hadoop fs -ls har:///archives/logs/2023/logs_2023_q1.har

# Step 4: Delete originals ONLY after validation
hadoop fs -rm -r /data/logs/2023/q1

# Expected: 1 HAR = 2 index files + N data files (vs. thousands of originals)
# Inode reduction ratio: typically 100:1 to 1000:1 for log workloads

<configuration>
  <!-- ViewFS: client-side namespace federation routing -->
  <property>
    <name>fs.defaultFS</name>
    <value>viewfs://ClusterX</value>
  </property>
  <property>
    <name>fs.viewfs.mounttable.ClusterX.link./data/raw</name>
    <value>hdfs://nn1:9000/data/raw</value>
  </property>
  <property>
    <name>fs.viewfs.mounttable.ClusterX.link./data/processed</name>
    <value>hdfs://nn2:9000/data/processed</value>
  </property>
  <property>
    <name>fs.viewfs.mounttable.ClusterX.link./data/archive</name>
    <value>hdfs://nn3:9000/data/archive</value>
  </property>
</configuration>

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count
import subprocess, logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("CompactionPipeline")

TARGET_FILE_BYTES  = 268_435_456   # 256 MB
MIN_FILE_THRESHOLD = 134_217_728   # compact partitions with avg file < 128 MB

def build_spark():
    return (SparkSession.builder
        .appName("SmallFileCompaction")
        .config("spark.sql.files.maxPartitionBytes",  "268435456")
        .config("spark.sql.shuffle.partitions",       "400")
        .config("spark.sql.adaptive.enabled",         "true")
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
        .config("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .getOrCreate())

def needs_compaction(spark, path):
    """Return True if avg file size < 128 MB or file count > 500."""
    fs_result = subprocess.run(
        ["hdfs", "dfs", "-count", path], capture_output=True, text=True)
    parts = fs_result.stdout.split()
    dirs, files, size_bytes = int(parts[0]), int(parts[1]), int(parts[2])
    avg = size_bytes // max(files, 1)
    log.info(f"  → {path}: {files} files, avg {avg/1e6:.1f} MB")
    return files > 500 or avg < MIN_FILE_THRESHOLD

def compact_partition(spark, src_path, tgt_path, fmt="parquet"):
    """Read, repartition to optimal count, write compressed output."""
    df = spark.read.format(fmt).load(src_path)
    total_bytes = spark.sparkContext._jvm.org.apache.hadoop \
        .fs.FileSystem.get(spark._jsc.hadoopConfiguration()) \
        .getContentSummary(
            spark.sparkContext._jvm.org.apache.hadoop.fs.Path(src_path)
        ).getLength()
    n_partitions = max(1, int(total_bytes / TARGET_FILE_BYTES))
    log.info(f"  → Writing {n_partitions} partitions to {tgt_path}")

    (df.coalesce(n_partitions)
       .write
       .mode("overwrite")
       .option("compression", "snappy")
       .format(fmt)
       .save(tgt_path))
    return n_partitions

def run_pipeline(root_path, partitions):
    spark = build_spark()
    for part in partitions:
        src = f"{root_path}/{part}"
        tmp = f"{root_path}/_compact_tmp/{part}"
        if not needs_compaction(spark, src):
            log.info(f"SKIP {part} — already optimal")
            continue
        log.info(f"COMPACT {part}")
        n = compact_partition(spark, src, tmp)
        # Atomic swap: rename tmp → src
        subprocess.run(["hdfs", "dfs", "-rm", "-r", src])
        subprocess.run(["hdfs", "dfs", "-mv", tmp, src])
        log.info(f"  ✓ Done. Output: {n} file(s).")
    spark.stop()

if __name__ == "__main__":
    # Drive from Airflow / cron; supply partition list from metadata scan
    run_pipeline("/data/events", ["dt=2024-01", "dt=2024-02", "dt=2024-03"])

-- Enable ACID and auto-compaction in hive-site.xml equivalent
SET hive.support.concurrency = true;
SET hive.exec.dynamic.partition.mode = nonstrict;
SET hive.compactor.initiator.on = true;
SET hive.compactor.worker.threads = 4;

-- Create optimal bucketed ORC table (avoids over-partitioning)
CREATE TABLE events_optimized (
  event_id     STRING,
  user_id      BIGINT,
  event_type   STRING,
  payload      STRING,
  event_ts     TIMESTAMP
)
PARTITIONED BY (event_date STRING)   -- coarse: date only, not hour
CLUSTERED BY (user_id) INTO 64 BUCKETS
STORED AS ORC
TBLPROPERTIES (
  'transactional'                        = 'true',
  'orc.compress'                         = 'SNAPPY',
  'orc.stripe.size'                      = '268435456',  -- 256 MB stripe
  'hive.compactor.minor.delta.num.threshold' = '10',
  'hive.compactor.major.size.threshold'  = '1073741824'  -- 1 GB
);

-- Manually trigger compaction on a specific partition
ALTER TABLE events_optimized PARTITION (event_date='2024-01-01')
  COMPACT 'MAJOR';

-- Check compaction queue status
SHOW COMPACTIONS;

# Ozone must be running alongside HDFS (Hadoop 3.2+ includes Ozone)

# Step 1: Initialize Ozone volume/bucket for cold archive
ozone sh volume create /vol-archive
ozone sh bucket create /vol-archive/cold-data --replication EC --ecData 6 --ecParity 3

# Step 2: DistCp cold partition to Ozone (runs as MapReduce, no extra HW)
hadoop distcp \
  -Dmapreduce.job.queuename=compaction \
  -bandwidth 100 \
  -m 50 \
  hdfs://nn1:9000/data/archive/logs/2022/ \
  o3fs://cold-data.vol-archive.ozone-service/logs/2022/

# Step 3: Verify checksum integrity
hadoop distcp \
  -Ddistcp.options.verify=true \
  -overwrite \
  hdfs://nn1:9000/data/archive/logs/2022/ \
  o3fs://cold-data.vol-archive.ozone-service/logs/2022/

# Step 4: Delete HDFS original after validation (reduces NN inodes)
hdfs dfs -rm -r /data/archive/logs/2022/

# Estimated throughput: ~500 MB/s across 50 mappers → 14 PB ≈ 8 days

-- Option A: Migrate in-place (Spark 3.x with Iceberg Spark extensions)
CALL catalog.system.migrate('default.events_legacy');

-- Option B: Create new Iceberg table and insert-overwrite
CREATE TABLE catalog.default.events_iceberg (
  event_id   STRING,
  user_id    BIGINT,
  event_type STRING,
  event_ts   TIMESTAMP
)
USING ICEBERG
PARTITIONED BY (days(event_ts))        -- hidden partitioning — no partition cols in data
TBLPROPERTIES (
  'write.format.default'           = 'parquet',
  'write.parquet.compression-codec' = 'zstd',
  'write.target-file-size-bytes'    = '268435456',    -- 256 MB target
  'write.distribution-mode'        = 'hash',
  'history.expire.min-snapshots-to-keep' = '5'
);

-- Compact Iceberg table (rewrite small files into optimal size)
CALL catalog.system.rewrite_data_files(
  table => 'default.events_iceberg',
  strategy => 'binpack',
  options => map(
    'target-file-size-bytes', '268435456',
    'min-input-files',        '5'
  )
);

-- Expire old snapshots to clean manifest bloat
CALL catalog.system.expire_snapshots(
  table => 'default.events_iceberg',
  older_than => TIMESTAMP '2024-01-01 00:00:00',
  retain_last => 7
);

from pyspark.sql import SparkSession

spark = (SparkSession.builder
  .config("spark.sql.extensions",
          "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
  .config("spark.sql.catalog.catalog",
          "org.apache.iceberg.spark.SparkCatalog")
  .config("spark.sql.catalog.catalog.type",           "hive")
  .config("spark.sql.catalog.catalog.uri",
          "thrift://hive-metastore:9083")
  .config("spark.sql.catalog.catalog.warehouse",
          "hdfs://nn1:9000/warehouse/iceberg")
  # Tune Iceberg write for large files
  .config("spark.sql.iceberg.planning.preserve-data-grouping", "true")
  .config("spark.sql.iceberg.handle-timestamp-without-timezone", "true")
  .getOrCreate())

-- Register Iceberg catalog in Doris (Multi-Catalog)
CREATE CATALOG iceberg_hdfs PROPERTIES (
  'type'                           = 'iceberg',
  'iceberg.catalog.type'           = 'hms',
  'hive.metastore.uris'            = 'thrift://hive-metastore:9083',
  'hadoop.fs.defaultFS'            = 'hdfs://nn1:9000',
  'dfs.nameservices'               = 'mycluster'
);

-- Query without touching NameNode LIST — metadata served from manifests
SELECT event_type, COUNT(*) AS cnt
FROM iceberg_hdfs.default.events_iceberg
WHERE event_ts >= '2024-01-01'
GROUP BY event_type
ORDER BY cnt DESC;