SOLUTION 1: HAR FILES (Hadoop Archive)
─────────────────────────────────────
  Groups many small files into one HAR archive
  Single metadata entry, but files accessible individually
  hadoop archive -archiveName bookings.har -p /logs/small/ /archives/

  Pros: Reduces NameNode metadata pressure
  Cons: Read-only (can't append), slower random access
  Use when: archiving old data you rarely need to process

SOLUTION 2: SEQUENCE FILES
─────────────────────────────────────
  Binary format: key-value pairs, multiple small files merged into one
  Key = filename, Value = file content
  Used in: MapReduce as input, Kafka → HDFS pipelines

  Pros: Splittable (supports parallel reads), fast
  Cons: Not human-readable, only MapReduce/Spark can read
  Use when: processing pipeline, not for human inspection

SOLUTION 3: COMBINE INPUT FORMAT (MapReduce)
─────────────────────────────────────
  CombineFileInputFormat groups multiple small files into one Map task
  Instead of 10,000 files = 10,000 map tasks:
  CombineFileInputFormat = 100 map tasks (each processes 100 files)

  In Hive: SET hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

SOLUTION 4: HIVE MERGE ON INSERT (most practical)
─────────────────────────────────────
  Hive creates small files on every INSERT INTO partition
  Solution: merge small files after write

  SET hive.merge.mapfiles=true;          -- merge after map-only jobs
  SET hive.merge.mapredfiles=true;       -- merge after map-reduce jobs
  SET hive.merge.size.per.task=256000000; -- target 256 MB per merged file
  SET hive.merge.smallfiles.avgsize=16000000; -- trigger if avg < 16 MB

SOLUTION 5: SPARK COALESCE/REPARTITION (modern approach)
─────────────────────────────────────
  After processing, write fewer, larger files:
  df.coalesce(10).write.parquet("/output/bookings/")
  -- Reduces from 10,000 tiny files to 10 files of ~128 MB each

  This is the MODERN solution — most interviewers expect this answer!

SOLUTION 6: AVOID CREATING SMALL FILES
─────────────────────────────────────
  Root cause fix: why are small files being created?
  - Streaming writes: use batching (Kafka → buffer 128 MB before writing)
  - Many small partitions: use fewer, coarser partitions
  - Dynamic partitions: too many partition values → too many files
    Solution: partition by month instead of day if data is sparse

# ═══════════════════════════════════
# FILE SYSTEM OPERATIONS
# ═══════════════════════════════════
hdfs dfs -ls /user/data/bookings/          # List directory contents
hdfs dfs -ls -R /user/data/                # List RECURSIVELY (all subdirs)
hdfs dfs -du -s -h /user/data/bookings/   # Disk usage (-h=human readable, -s=summary)
hdfs dfs -mkdir -p /user/data/2026/march/ # Create directory (and parents with -p)
hdfs dfs -put bookings.csv /user/data/    # Upload local file to HDFS
hdfs dfs -get /user/data/bookings.csv .   # Download from HDFS to local
hdfs dfs -cat /user/data/file.csv         # Print file contents to stdout
hdfs dfs -tail /user/data/logs.txt        # Print last 1 KB of file (like Unix tail)
hdfs dfs -cp /src/file.csv /dst/          # Copy within HDFS
hdfs dfs -mv /src/file.csv /dst/          # Move within HDFS
hdfs dfs -rm /user/data/old.csv           # Delete file (goes to Trash by default!)
hdfs dfs -rm -r /user/data/old_dir/      # Delete directory recursively
hdfs dfs -rm -skipTrash /user/data/old.csv  # Delete PERMANENTLY (bypass Trash)
hdfs dfs -expunge                          # Empty Trash (permanently delete)
hdfs dfs -setrep -R 2 /user/data/archive/ # Change replication factor to 2 (save space)
hdfs dfs -checksum /user/data/file.csv    # Get MD5/CRC checksum of file

# ═══════════════════════════════════
# ADMIN OPERATIONS
# ═══════════════════════════════════
hdfs dfsadmin -report                     # Cluster status (live nodes, capacity, used)
hdfs dfsadmin -safemode get               # Check if HDFS is in safe mode
hdfs dfsadmin -safemode leave             # Force exit safe mode (after NameNode restart)
hdfs dfsadmin -refreshNodes               # Re-read includes/excludes (add/remove DNs)
hdfs dfsadmin -setQuota 100 /user/team1/ # Set namespace quota (max 100 files)
hdfs dfsadmin -setSpaceQuota 1t /user/team1/ # Set storage quota (1 TB)

# ═══════════════════════════════════
# FSCK — Filesystem health check
# ═══════════════════════════════════
hdfs fsck /user/data/bookings/            # Check health of all files
hdfs fsck / -files -blocks -locations     # Show all files + block locations
hdfs fsck / -list-corruptfileblocks       # List corrupted blocks
# Output to watch for:
# "HEALTHY" = all good
# "Under-replicated blocks" = some blocks don't have 3 replicas (DataNode down?)
# "Missing blocks" = data loss! Block has 0 replicas (need to recover from backup)
# "Corrupt blocks" = checksum mismatch (need to restore from another replica)

# ═══════════════════════════════════
# BALANCER — Fix data skew across DataNodes
# ═══════════════════════════════════
hdfs balancer -threshold 10               # Balance cluster (max 10% imbalance)
# Run when: adding new DataNodes (they start empty, all writes go to them)
# Balancer moves blocks from full nodes to empty nodes gradually

# ═══════════════════════════════════
# SAFE MODE — Important to understand
# ═══════════════════════════════════
# Safe mode = NameNode startup state
# During safe mode: HDFS is READ-ONLY (no writes allowed)
# NameNode waits for DataNodes to report their blocks
# Safe mode exits when: enough blocks have minimum replication
# ⚠️ If NameNode is stuck in safe mode → cluster appears DOWN
# Fix: hdfs dfsadmin -safemode leave  (only if blocks are actually replicated!)

<!-- mapred-site.xml — MapReduce memory config -->

<!-- Container memory for Map tasks -->
<property>
  <name>mapreduce.map.memory.mb</name>
  <value>2048</value>
  <!-- Default is 1024 MB. Increase if mappers are getting killed (OOM) -->
  <!-- Rule: set to 1.5x - 2x your data per mapper -->
</property>

<!-- Container memory for Reduce tasks -->
<property>
  <name>mapreduce.reduce.memory.mb</name>
  <value>4096</value>
  <!-- Reducers need more memory than Mappers (they hold sorted data) -->
  <!-- Rule: 2x mapper memory is a safe starting point -->
</property>

<!-- JVM heap for Map tasks (must be LESS than container memory) -->
<property>
  <name>mapreduce.map.java.opts</name>
  <value>-Xmx1638m</value>
  <!-- Rule: 80% of mapreduce.map.memory.mb -->
  <!-- 2048 * 0.8 = 1638 MB — remaining 20% is for JVM overhead -->
</property>

<!-- JVM heap for Reduce tasks -->
<property>
  <name>mapreduce.reduce.java.opts</name>
  <value>-Xmx3276m</value>
  <!-- Rule: 80% of mapreduce.reduce.memory.mb = 4096 * 0.8 = 3276 MB -->
</property>

<!-- The sort buffer is IN-MEMORY — bigger = fewer disk spills = FASTER shuffle -->

<property>
  <name>mapreduce.task.io.sort.mb</name>
  <value>512</value>
  <!-- Default: 100 MB — way too small for large jobs -->
  <!-- Increase to 512 MB or more to reduce disk spills during shuffle -->
  <!-- Must fit within map container memory (mapreduce.map.memory.mb) -->
  <!-- Tuning: If you see "Spilling to disk" logs → increase this first! -->
</property>

<!-- When buffer is X% full, start spilling to disk (while still mapping) -->
<property>
  <name>mapreduce.map.sort.spill.percent</name>
  <value>0.80</value>
  <!-- Default: 0.80 (80%) — spill when 80% of sort buffer is used -->
  <!-- Increasing to 0.90 keeps more in memory but risks OOM -->
</property>

<!-- Number of parallel sort/merge threads for spill files -->
<property>
  <name>mapreduce.task.io.sort.factor</name>
  <value>100</value>
  <!-- Default: 10 — how many spill files merged at once -->
  <!-- Increase to reduce the number of merge passes → fewer I/O operations -->
</property>

<!-- Compress map output (the shuffle data — this crosses network!) -->
<property>
  <name>mapreduce.map.output.compress</name>
  <value>true</value>
  <!-- TRUE = compress data during shuffle → HUGE network bandwidth savings -->
  <!-- Cost: CPU for compress/decompress (usually worth it on network-bound jobs) -->
</property>

<property>
  <name>mapreduce.map.output.compress.codec</name>
  <value>org.apache.hadoop.io.compress.SnappyCodec</value>
  <!-- Snappy: FAST compress/decompress, moderate compression ratio -->
  <!-- Best choice for intermediate shuffle data (speed matters here) -->
  <!-- Alternatives: LZ4 (even faster), Zlib (smaller but slower) -->
</property>

<!-- Compress final job output -->
<property>
  <name>mapreduce.output.fileoutputformat.compress</name>
  <value>true</value>
</property>

<property>
  <name>mapreduce.output.fileoutputformat.compress.codec</name>
  <value>org.apache.hadoop.io.compress.GzipCodec</value>
  <!-- GZIP for final output: better compression ratio (smaller files stored long-term) -->
  <!-- ⚠️ GZIP is NOT splittable! Use LZO or Snappy if next job needs to read this -->
  <!-- LZO: splittable + fast — best for output that will be input to another MR job -->
  <!-- Bzip2: most compressed, splittable, but VERY slow compression -->
</property>

<!-- Number of reduce tasks (parallel reducers) -->
<property>
  <name>mapreduce.job.reduces</name>
  <value>10</value>
  <!-- Default: 1 (terrible for large jobs!) -->
  <!-- Rule of thumb: 0.95 × (nodes × max_containers_per_node) -->
  <!-- Example: 10 nodes × 4 containers = 40 → set reducers = 38 -->
  <!-- Too many reducers: overhead of managing many tasks + small output files -->
  <!-- Too few reducers: bottleneck, data skew impact is worse -->
  <!-- 0 reducers: no shuffle phase (map-only job) — for filter/transform jobs! -->
</property>

<!-- Reuse JVM containers across multiple mapper tasks -->
<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>
  <!-- Default: 1 (new JVM per task = slow startup for each task!) -->
  <!-- Setting to 10: same JVM handles 10 tasks before restarting -->
  <!-- Setting to -1: unlimited reuse (one JVM per slot for entire job) -->
  <!-- HUGE gain when you have MANY small mapper tasks (small files problem!) -->
  <!-- ⚠️ Can cause memory leaks if tasks have bugs → use with stable code only -->
</property>

<!-- Automatically re-run slow (straggler) tasks on another node -->
<property>
  <name>mapreduce.map.speculative</name>
  <value>true</value>
  <!-- Default: true — Hadoop automatically detects slow mappers and duplicates them -->
</property>

<property>
  <name>mapreduce.reduce.speculative</name>
  <value>true</value>
  <!-- Default: true -->
  <!-- ⚠️ Disable if your tasks have side effects (writing to DB, calling API) -->
  <!-- Otherwise same row could be written TWICE to the external system! -->
</property>

# Set block size when writing a specific file (512 MB blocks)
hdfs dfs -D dfs.blocksize=536870912 -put large_file.dat /data/archive/

# Or set in your application
# In MapReduce job config:
job.getConfiguration().set("dfs.blocksize", "536870912");

<!-- yarn-site.xml — resource manager settings -->

<!-- Total memory available for YARN on each NodeManager -->
<property>
  <name>yarn.nodemanager.resource.memory-mb</name>
  <value>49152</value>
  <!-- On a 64 GB node: reserve 8 GB for OS + other services = 56 GB, then YARN gets 48 GB -->
  <!-- Rule: YARN memory = total RAM - OS overhead (8-16 GB) - HBase/etc memory -->
</property>

<!-- Total CPU vcores available for YARN on each NodeManager -->
<property>
  <name>yarn.nodemanager.resource.cpu-vcores</name>
  <value>16</value>
  <!-- On a 16-core node: give all 16 to YARN (OS uses very little normally) -->
</property>

<!-- Minimum container size (floor) -->
<property>
  <name>yarn.scheduler.minimum-allocation-mb</name>
  <value>512</value>
  <!-- No container gets less than 512 MB — prevents tiny useless containers -->
</property>

<!-- Maximum container size (ceiling) -->
<property>
  <name>yarn.scheduler.maximum-allocation-mb</name>
  <value>16384</value>
  <!-- No single container gets more than 16 GB (prevents one job hogging all RAM) -->
</property>

<!-- Minimum vcores per container -->
<property>
  <name>yarn.scheduler.minimum-allocation-vcores</name>
  <value>1</value>
</property>

<!-- Maximum vcores per container -->
<property>
  <name>yarn.scheduler.maximum-allocation-vcores</name>
  <value>4</value>
</property>

<!-- capacity-scheduler.xml -->

<!-- Define queue capacities (must add up to 100%) -->
<property>
  <name>yarn.scheduler.capacity.root.production.capacity</name>
  <value>70</value>
  <!-- Production queue gets 70% of cluster resources guaranteed -->
</property>

<property>
  <name>yarn.scheduler.capacity.root.development.capacity</name>
  <value>20</value>
</property>

<property>
  <name>yarn.scheduler.capacity.root.adhoc.capacity</name>
  <value>10</value>
</property>

<!-- Maximum capacity (can use idle resources from other queues) -->
<property>
  <name>yarn.scheduler.capacity.root.production.maximum-capacity</name>
  <value>90</value>
  <!-- Production can burst to 90% if other queues are idle -->
</property>

<!-- User limit: how much one user can use within a queue -->
<property>
  <name>yarn.scheduler.capacity.root.production.user-limit-factor</name>
  <value>2</value>
  <!-- One user can use up to 2x their fair share within the queue -->
</property>

# Change replication factor for cold/archive data (save storage)
hdfs dfs -setrep -w 1 /data/archive/2020/     # Old archive → 1 copy (risk: data loss!)
hdfs dfs -setrep -w 2 /data/historical/2023/  # Historical → 2 copies (balanced)
hdfs dfs -setrep -w 3 /data/production/       # Hot production data → 3 copies (default)

# Check current replication factor of a file
hdfs fsck /data/archive/2020/ -files | grep "replication"

# HDFS Balancer — rebalance blocks across DataNodes (run after adding new nodes)
hdfs balancer -threshold 10
# threshold: consider node "balanced" if within 10% of average utilization
# Run this AFTER adding new DataNodes — new nodes start empty, data doesn't auto-move!

# Or run balancer as background process
hdfs balancer -threshold 5 &

HIVE ARCHITECTURE = "DMCTE"
    D — Driver (receives SQL queries, manages lifecycle)
    M — Metastore (schema + HDFS location mapping — the catalog)
    C — Compiler (parses SQL → query plan)
    T — Tez/MR Engine (executes the plan)
    E — Execution (launches tasks on YARN)

HIVE OPTIMIZATION = "VECTOR-TOP":
    V — Vectorization (process 1024 rows at once, not 1)
    E — Engine: evaluate Tez instead of MapReduce for multi-stage DAG execution
    C — Column pruning (SELECT only needed columns)
    T — Table stats: ANALYZE TABLE for CBO
    O — ORC format (columnar + predicate pushdown)
    R — REPARTITION/REDISTRIBUTE (avoid data skew)

    T — Tez DAG (avoid MapReduce chains)
    O — Order of joins (small table FIRST for map join)
    P — Partition pruning (WHERE on partition columns only)

HIVE FILE FORMATS = "OPTA"
    O — ORC    (best for Hive: columnar, ACID, compression)
    P — Parquet (best for cross-tool: Spark, Impala, Hive)
    T — Text   (default, row-oriented; compressed input supported; useful for landing/interchange)
    A — Avro   (row-based, best for Kafka/Sqoop schema evolution)

PARTITIONING RULES = "QDLF"
    Q — Query filter columns → partition by those columns
    D — Date/region usually best partition keys
    L — Low cardinality (10s to 100s of values, not millions!)
    F — Few files per partition (avoid small files inside partitions)

HOW HIVE WORKS (end-to-end):

You type:  SELECT country, COUNT(*) FROM bookings GROUP BY country;
           ↓
Driver:    Receives your query, manages session
           ↓
Compiler:  Parses SQL → AST (Abstract Syntax Tree)
           ↓
Metastore: Compiler asks Metastore: "where is 'bookings' table in HDFS?"
           → Returns: /user/hive/warehouse/bookings, format=ORC, partitions=[2023,2024]
           ↓
Optimizer: Rewrite query to most efficient plan (Cost-Based Optimizer if stats exist)
           → Decides: map join or shuffle join? apply partition pruning?
           ↓
Execution: Submit MapReduce/Tez job to YARN
           ↓
Output:    Results written to HDFS or returned to client

INTERNAL TABLE (Managed Table):
    CREATE TABLE bookings (
        booking_id STRING,
        passenger STRING,
        flight STRING,
        amount DOUBLE
    )
    STORED AS ORC;
    -- Data goes to: /user/hive/warehouse/bookings/
    -- Hive OWNS this data

    WHAT HAPPENS ON DROP TABLE:
    → DROP TABLE bookings;
    → Hive deletes BOTH metadata (from Metastore) AND data (from HDFS)!
    ⚠️ Data is GONE. Cannot recover unless Trash is enabled.

EXTERNAL TABLE:
    CREATE EXTERNAL TABLE bookings_external (
        booking_id STRING,
        passenger STRING
    )
    LOCATION '/data/raw/bookings/'   -- You specify HDFS path
    STORED AS ORC;
    -- Hive points TO this data but does NOT own it

    WHAT HAPPENS ON DROP TABLE:
    → DROP TABLE bookings_external;
    → Hive deletes ONLY metadata from Metastore
    → Data in /data/raw/bookings/ is UNTOUCHED ← this is the key difference!

USE INTERNAL WHEN:
  ✓ Hive is the only tool accessing this data
  ✓ You want Hive to manage lifecycle (cleanup = just DROP TABLE)
  ✓ Gold/final tables that won't be shared

USE EXTERNAL WHEN:
  ✓ Data is shared with Spark, Presto, Pig, other tools
  ✓ Data is ingested by another process (Sqoop, Flume, Kafka)
  ✓ You CANNOT afford accidental data deletion on DROP TABLE
  ✓ Data lives outside /user/hive/warehouse/
  ✓ Production Bronze/Silver tables (raw landing, shared data)

EXAMPLE:
  Raw flight data ingested by Sqoop every hour → EXTERNAL TABLE
  (Sqoop writes to HDFS path, Hive just reads it, never owns it)
  Aggregated booking report table → INTERNAL TABLE
  (Hive fully manages this output)

-- Partitioned table: booking_date is the partition column
-- booking_date does NOT appear in the regular column list!
CREATE TABLE bookings (
    booking_id  STRING,
    passenger   STRING,
    flight_code STRING,
    amount      DOUBLE
)
PARTITIONED BY (booking_date STRING)  -- partition key goes here
STORED AS ORC;

-- This creates directory structure:
-- /user/hive/warehouse/bookings/booking_date=2024-01-15/part-00000.orc
-- /user/hive/warehouse/bookings/booking_date=2024-01-16/part-00000.orc
-- ...

-- STATIC PARTITIONING: you manually specify the partition value
INSERT INTO bookings PARTITION (booking_date='2024-01-15')
SELECT booking_id, passenger, flight_code, amount
FROM bookings_raw
WHERE booking_date = '2024-01-15';
-- Loads ONLY one partition at a time
-- SAFE: you control exactly which partition is written
-- Use for: loading historical data one date at a time

-- DYNAMIC PARTITIONING: Hive reads the partition value from the data itself
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
-- nonstrict = allow ALL partitions to be dynamic (at least 1 static needed for strict mode)

INSERT INTO bookings PARTITION (booking_date)
SELECT booking_id, passenger, flight_code, amount, booking_date  -- booking_date last!
FROM bookings_raw;
-- Hive reads booking_date column from each row, creates partition automatically
-- ⚠️ Can create HUNDREDS of partitions in one query → many small files problem!
-- ⚠️ If booking_date has 1000 unique dates → 1000 partitions × reducers = MANY files

-- Limit max dynamic partitions (safety net)
SET hive.exec.max.dynamic.partitions=1000;
SET hive.exec.max.dynamic.partitions.pernode=256;
-- If query would create more than 1000 partitions → FAIL (prevents cluster overload)

-- Merge small output files after insert
SET hive.merge.mapredfiles=true;
SET hive.merge.mapfiles=true;
SET hive.merge.smallfiles.avgsize=128000000;  -- 128 MB target
SET hive.merge.size.per.task=256000000;       -- 256 MB max
-- After INSERT, Hive runs a merge job to combine small output files into larger ones
-- This prevents small files inside partitions!

-- THE SCENARIO: Files added directly to HDFS (bypassing Hive)
-- Example: Sqoop wrote data to /user/hive/warehouse/bookings/booking_date=2024-02-01/
-- But Hive Metastore doesn't know this partition exists!

-- SYMPTOM:
SELECT * FROM bookings WHERE booking_date = '2024-02-01';
-- Returns: 0 rows (Metastore says this partition doesn't exist!)

-- FIX:
MSCK REPAIR TABLE bookings;
-- "MSCK" = MetaStore Check
-- What it does: scans HDFS for partition directories, adds missing partitions to Metastore
-- After this: Hive knows about the new partition → query works!

-- FASTER ALTERNATIVE (add one specific partition):
ALTER TABLE bookings ADD PARTITION (booking_date='2024-02-01')
LOCATION '/user/hive/warehouse/bookings/booking_date=2024-02-01/';

-- ⚠️ MSCK REPAIR can be SLOW on tables with THOUSANDS of partitions
-- (scans all of HDFS under the table path)
-- For large tables: use ALTER TABLE ADD PARTITION instead!

-- Create bucketed table: divide by booking_id into 32 buckets
CREATE TABLE bookings_bucketed (
    booking_id  STRING,
    passenger   STRING,
    amount      DOUBLE
)
CLUSTERED BY (booking_id) INTO 32 BUCKETS
STORED AS ORC;

-- Insert data — Hive automatically routes rows to correct bucket via hash(booking_id) % 32
INSERT INTO bookings_bucketed SELECT * FROM bookings_raw;

-- Bucket JOIN optimization (two tables bucketed by SAME column in SAME number of buckets)
-- Bucket 1 from table A joins with Bucket 1 from table B → NO SHUFFLE!
SET hive.optimize.bucketmapjoin=true;
SET hive.optimize.bucketmapjoin.sortedmerge=true;

-- Hash sampling into 10 logical sample buckets:
SELECT * FROM bookings_bucketed TABLESAMPLE(BUCKET 1 OUT OF 10 ON booking_id);
-- This deterministically selects one hash bucket, but does not guarantee exactly 10% of rows.
-- Physical input pruning aligns best when y divides or is a multiple of the table's bucket count.
-- Here 10 is neither a divisor nor a multiple of 32, so Hive cannot map the sample cleanly
-- to whole physical bucket files and may need to scan broader input.

-- Alignment example for a 32-bucket table: y=16 selects two physical buckets.
SELECT * FROM bookings_bucketed TABLESAMPLE(BUCKET 1 OUT OF 16 ON booking_id);

-- Create ORC table with compression:
CREATE TABLE bookings_orc (
    booking_id STRING, passenger STRING, amount DOUBLE
)
STORED AS ORC
TBLPROPERTIES ("orc.compress"="SNAPPY");
-- SNAPPY = fast (good for hot frequently-read data)
-- ZLIB = more compressed (good for cold archive data)

-- Create Parquet table:
CREATE TABLE bookings_parquet (
    booking_id STRING, passenger STRING, amount DOUBLE
)
STORED AS PARQUET;

-- Convert existing Text table to ORC (HUGE performance win):
CREATE TABLE bookings_orc STORED AS ORC AS
SELECT * FROM bookings_text;
-- This can materially reduce scanned bytes; measure the full query rather than assuming a multiplier.

-- DEFAULT in older clusters: MapReduce (slow)
-- SET this at the START of every Hive session:
SET hive.execution.engine=tez;  -- Or 'mr' for MapReduce (don't use)

-- WHY TEZ IS FASTER:
-- MapReduce: Chain of MR jobs → each writes to HDFS between stages
-- SELECT country, COUNT(*), SUM(amount) GROUP BY country ORDER BY country
-- MapReduce: Job1 (GROUP BY) → write HDFS → Job2 (ORDER BY) → write HDFS
-- Tez:       One DAG: GROUP BY → ORDER BY; it avoids materializing every MR stage as HDFS output, but can still spill.
-- Result: often less startup and intermediate-I/O overhead; benchmark the actual plan.

-- Enable vectorized query execution
SET hive.vectorized.execution.enabled=true;
SET hive.vectorized.execution.reduce.enabled=true;

-- WHAT IT DOES:
-- Default: process 1 row at a time (row-by-row)
-- Vectorized: process 1024 rows at once (batch)
-- Uses CPU SIMD instructions (process multiple values in parallel on the CPU)
-- Result: batch execution can improve CPU efficiency for supported types and formats; measure it.
-- Works BEST with ORC format (which stores data column-by-column)
-- ⚠️ Only works with ORC (not Text or CSV tables)

-- Enable Cost-Based Optimizer
SET hive.cbo.enable=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;

-- CBO needs TABLE STATISTICS to work:
ANALYZE TABLE bookings COMPUTE STATISTICS;           -- row count, file sizes
ANALYZE TABLE bookings COMPUTE STATISTICS FOR COLUMNS booking_id, country, amount;
-- Column stats: min, max, distinct count, nulls
-- CBO uses these stats to choose optimal join order and join type

-- WITHOUT stats: Hive guesses → wrong join order → shuffle join for a small table
-- WITH stats:    CBO says "passengers table has 10k rows → use MAP JOIN, no shuffle"

-- MAP JOIN = load small table into EVERY mapper's RAM, join locally (NO SHUFFLE!)
-- Auto map join (CBO decides based on table size):
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=25000000;  -- 25 MB threshold
-- If the smaller table < 25 MB → Hive automatically uses map join
-- ⚠️ Increase this if you have slightly larger but still "small" dimension tables

-- FORCE map join for a specific query with hint:
SELECT /*+ MAPJOIN(airports) */
    b.booking_id,
    a.airport_name
FROM bookings b
JOIN airports a ON b.departure_airport = a.code;
-- airports table is small (100 airports) → broadcast to all mappers → NO shuffle!

-- BUCKET MAP JOIN (both tables bucketed by join key):
SET hive.optimize.bucketmapjoin=true;
-- Even faster: only load matching bucket from small table into mapper

-- PARTITION PRUNING: Hive SKIPS partitions not matching WHERE clause
-- ⚠️ ONLY WORKS if you filter on the PARTITION COLUMN!

-- ✓ GOOD — partition pruning works (only scans booking_date=2024-01-15):
SELECT COUNT(*) FROM bookings WHERE booking_date = '2024-01-15';

-- ✗ BAD — NO partition pruning (scans ALL partitions!):
SELECT COUNT(*) FROM bookings WHERE YEAR(booking_date) = 2024;
-- Hive applies function YEAR() to each row → can't prune without scanning all!
-- FIX: WHERE booking_date BETWEEN '2024-01-01' AND '2024-12-31'

-- ✗ ANOTHER TRAP — dynamic partition filter from subquery:
-- This may or may not prune depending on Hive version and optimizer
SELECT * FROM bookings b
JOIN date_dim d ON b.booking_date = d.date
WHERE d.year = 2024;
-- May not prune! Use: WHERE b.booking_date >= '2024-01-01'

-- Data skew in GROUP BY (one key dominates):
-- Example: 50% of bookings are from 'US' → one reducer handles all US data!

SET hive.groupby.skewindata=true;
-- What Hive does: 2-phase aggregation:
-- Phase 1: randomly distribute keys → partial aggregation in parallel
-- Phase 2: final aggregation by real key
-- Result: no single reducer bottleneck for skewed keys

-- Also for skewed JOIN:
SET hive.optimize.skewjoin=true;
SET hive.skewjoin.key=100000;  -- Keys with > 100k rows are treated as "skew keys"

-- ORDER BY vs SORT BY vs DISTRIBUTE BY vs CLUSTER BY:

ORDER BY column     -- Global sort (all data goes to ONE reducer) → SLOW for large data
SORT BY column      -- Local sort (each reducer sorts its own output) → FAST but not globally sorted
DISTRIBUTE BY col   -- Control which rows go to which reducer (like partitioner)
CLUSTER BY col      -- = DISTRIBUTE BY col SORT BY col (same column for both)

-- ⚡ INTERVIEW TRAP: "What's the difference between ORDER BY and SORT BY?"
-- ORDER BY: guaranteed global order, uses 1 reducer (slow for billions of rows!)
-- SORT BY: each reducer output is sorted, but globally NOT sorted (multiple reducers)
-- Use SORT BY when you just need local sorted chunks (e.g., before writing partitioned output)
-- Use ORDER BY only for final result sets that need to be globally ordered

-- WHAT HAPPENED:
-- Someone ran: hdfs dfs -rm -r /user/hive/warehouse/bookings/booking_date=2024-01-15/
-- Now this query returns 0 rows:
SELECT COUNT(*) FROM bookings WHERE booking_date = '2024-01-15';
-- Even though Metastore thinks the partition exists!

-- DIAGNOSIS:
-- 1. Check if partition exists in Metastore:
SHOW PARTITIONS bookings;
-- Shows: booking_date=2024-01-15 (it's there in Metastore!)
-- 2. Check if files exist in HDFS:
-- hdfs dfs -ls /user/hive/warehouse/bookings/booking_date=2024-01-15/
-- Output: No such file or directory ← CONFIRMED — data deleted!

-- FIX OPTIONS:
-- Option 1: Restore data from backup and MSCK REPAIR TABLE bookings
-- Option 2: If no backup — DROP the stale partition from Metastore:
ALTER TABLE bookings DROP PARTITION (booking_date='2024-01-15');
-- Now Hive is consistent again (it won't show non-existent partition)
-- Then: reload data into that partition from source

-- PREVENTION:
-- 1. NEVER run hdfs dfs -rm directly on Hive-managed table locations
-- 2. Use HDFS Trash (hdfs dfs -rm moves to .Trash, not permanent delete)
-- 3. Enable HDFS Snapshots for the warehouse directory

-- WHAT HAPPENED: Sqoop/Spark wrote new partition directory to HDFS
-- but forgot to register it in Hive Metastore

-- DIAGNOSIS:
SELECT * FROM bookings WHERE booking_date = '2024-03-01';
-- Returns 0 rows... but file exists in HDFS!

-- FIX — Option 1: MSCK REPAIR (for EXTERNAL tables, scans all partitions):
MSCK REPAIR TABLE bookings;

-- FIX — Option 2: ADD PARTITION (fast, for specific known partition):
ALTER TABLE bookings ADD IF NOT EXISTS PARTITION (booking_date='2024-03-01')
LOCATION '/user/hive/warehouse/bookings/booking_date=2024-03-01/';

-- ⚠️ MSCK REPAIR on a table with 10,000 partitions → scans 10,000 HDFS paths!
-- Use ALTER TABLE ADD PARTITION when you know exactly which partition to add

-- STEP 1: Enable EXPLAIN to see query plan:
EXPLAIN SELECT country, COUNT(*) FROM bookings GROUP BY country;
-- Look for: Map Join vs Shuffle Join, partition pruning status, # of Map/Reduce stages

-- STEP 2: Check query stats after run:
SET hive.stats.autogather=true;
-- After INSERT: Hive auto-collects stats (rows, bytes)

-- STEP 3: Check if vectorization is working:
EXPLAIN VECTORIZATION SELECT country, COUNT(*) FROM bookings_orc GROUP BY country;
-- Should show "Vectorized execution: true" for ORC tables

-- STEP 4: Check for data skew:
-- If one reducer takes much longer than others → skew problem
SET hive.groupby.skewindata=true;

-- STEP 5: Confirm Tez engine:
SET hive.execution.engine;
-- Should show: tez (if mr → change to tez immediately!)

HBASE = NoSQL column-family database on top of HDFS

WHY HBASE? HDFS is write-once (append-only). HBase gives you random read/write!
Use when:
  ✓ Need random row lookups by key (e.g., GET booking_id = 'B12345')
  ✓ Need sub-10 ms response time for individual record access
  ✓ Millions of concurrent point queries
  ✓ Wide-column schema (each row can have different columns)

HBASE vs HIVE:
  HBase: real-time point lookups (GET/PUT/DELETE by row key)
  Hive:  batch analytics (SELECT COUNT(*) GROUP BY)
  HBase: low-latency random access when row keys and regions are designed well; Hive: analytic scans
  HBase: NoSQL, Hive: SQL
  Both: run on top of HDFS

# BASIC IMPORT: Oracle table → HDFS
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/ENTERPRISE \
  --username etl_user \
  --password 'secret' \
  --table FLIGHT_INVENTORY \
  --target-dir /data/raw/flight_inventory/ \
  --num-mappers 4             # 4 parallel import tasks

# IMPORT WITH HIVE INTEGRATION: directly to Hive table
sqoop import \
  --connect jdbc:mysql://db-host/bookings \
  --username user --password pass \
  --table bookings \
  --hive-import \             # create Hive table automatically
  --hive-table booking_db.bookings_raw \
  --create-hive-table \
  --num-mappers 8

# INCREMENTAL IMPORT (most important for production!):
# Only import NEW rows since last run (much faster than full import daily)

# Mode 1: append — for insert-only tables (new rows have higher IDs)
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/ENTERPRISE \
  --table BOOKINGS \
  --incremental append \
  --check-column booking_id \     # column to check for new rows
  --last-value 1000000 \          # only import booking_id > 1000000
  --target-dir /data/raw/bookings_incremental/

# Mode 2: lastmodified — for tables with an update timestamp
sqoop import \
  --connect jdbc:oracle:thin:@//db-host:1521/ENTERPRISE \
  --table BOOKINGS \
  --incremental lastmodified \
  --check-column updated_at \     # timestamp column
  --last-value "2024-01-15 00:00:00" \
  --target-dir /data/raw/bookings_incremental/

# --split-by: control how data is split across mappers (affects parallelism)
sqoop import \
  --table BOOKINGS \
  --split-by booking_date \       # each mapper handles a date range
  --num-mappers 8
# ⚠️ Default: split by primary key (must be numeric!)
# ⚠️ If no primary key and no --split-by → must use --num-mappers 1 (slow!)

<!-- workflow.xml — simple Hive workflow -->
<workflow-app name="daily-booking-report" xmlns="uri:oozie:workflow:0.5">

    <start to="hive-action"/>

    <!-- Action 1: Run Hive query -->
    <action name="hive-action">
        <hive xmlns="uri:oozie:hive-action:0.5">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <script>/user/oozie/scripts/aggregate_bookings.hql</script>
            <param>booking_date=${booking_date}</param>
        </hive>
        <ok to="sqoop-export"/>
        <error to="fail"/>
    </action>

    <!-- Action 2: Export result to Oracle -->
    <action name="sqoop-export">
        <sqoop xmlns="uri:oozie:sqoop-action:0.4">
            <command>export --connect jdbc:oracle:... --table REPORT_TABLE ...</command>
        </sqoop>
        <ok to="end"/>
        <error to="fail"/>
    </action>

    <kill name="fail">
        <message>Workflow failed: ${wf:errorMessage(wf:lastErrorNode())}</message>
    </kill>

    <end name="end"/>
</workflow-app>

HIVE ARCHITECTURE = "DMCTE"
    D — Driver (receives SQL, manages lifecycle)
    M — Metastore (schema + HDFS location → MySQL backend!)
    C — Compiler (SQL → query plan)
    T — Tez/MR engine (executes plan on YARN)
    E — Execution (submits to YARN, reads from HDFS)

HIVE OPTIMIZATION = "VECTOR-TOP":
    V — Vectorization (1024 rows at once: hive.vectorized.execution.enabled=true)
    E — Engine=Tez (hive.execution.engine=tez, not mr)
    C — CBO stats (ANALYZE TABLE + hive.cbo.enable=true)
    T — Tez DAG (can pipeline stages and avoid mandatory HDFS materialization between every stage)
    O — ORC format (columnar + predicate pushdown + ACID)
    R — REPARTITION/skew fix (hive.groupby.skewindata=true)

    T — Table map join (hive.auto.convert.join=true, small table → broadcast)
    O — ORDER BY → SORT BY (local sort, use instead of global sort)
    P — Partition pruning (filter on partition column, not derived expressions)

HIVE FILE FORMATS = "OPTA":
    O — ORC    (Hive-native, ACID, predicate pushdown — BEST FOR HIVE)
    P — Parquet (cross-tool: Spark+Impala+Hive — BEST FOR MULTI-TOOL)
    T — Text   (default row format; compressed input supported; best for landing/interchange)
    A — Avro   (schema evolution, Kafka/Sqoop landing — ROW-based)

INTERNAL vs EXTERNAL:
    Internal = Hive OWNS data → DROP TABLE = data DELETED from HDFS
    External = Hive POINTS to data → DROP TABLE = only metadata deleted, data SAFE
    ⚡ Rule: External for shared/raw data, Internal for Hive-only output

ECOSYSTEM = "SQOF-ZK":
    S — Sqoop (RDBMS ↔ HDFS transfer)
    Q — (Hive = SQL on HDFS)
    O — Oozie (workflow + coordinator jobs)
    F — Flume (log streaming → HDFS)
    ZK — ZooKeeper (distributed coordination, leader election)

┌──────────────────────────────────────────────────────────────────┐
│                  DAY 2: HIVE + ECOSYSTEM                          │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  HIVE ARCHITECTURE = "DMCTE":                                    │
│  Driver → Metastore → Compiler → Tez Engine → Execution          │
│  Metastore backend = MySQL (NOT HDFS) — it goes down → all fail  │
│                                                                  │
│  INTERNAL vs EXTERNAL:                                           │
│  Internal: DROP TABLE = metadata + HDFS data deleted!            │
│  External: DROP TABLE = only metadata, HDFS data SAFE            │
│  Use External for: raw data, shared data, Sqoop imports          │
│                                                                  │
│  PARTITIONING:                                                   │
│  Static: PARTITION (date='2024-01-15') — manual                  │
│  Dynamic: SET hive.exec.dynamic.partition.mode=nonstrict         │
│  MSCK REPAIR TABLE = add missing partitions to Metastore         │
│  Partition pruning: filter DIRECTLY on partition column!         │
│                                                                  │
│  HIVE OPTIMIZATION = "VECTOR-TOP":                               │
│  1. Engine=Tez (5-10x vs MapReduce)                              │
│  2. Vectorization (1024 rows batch, ORC only)                    │
│  3. CBO + ANALYZE TABLE (stats for optimal join plan)            │
│  4. Map Join (small table < 25 MB → broadcast, no shuffle)        │
│  5. ORC + predicate pushdown (skip stripes by min/max)           │
│  6. skewindata=true (2-phase GROUP BY for hot keys)              │
│  7. SORT BY not ORDER BY (unless global sort needed)             │
│                                                                  │
│  FILE FORMATS = "OPTA":                                          │
│  ORC: Hive-native, ACID, predicate pushdown                      │
│  Parquet: cross-tool (Spark+Impala+Hive)                         │
│  Text: landing/interchange; compressed input supported           │
│  Avro: schema evolution, Kafka/Sqoop landing                     │
│                                                                  │
│  HBASE:                                                          │
│  Random read/write on HDFS (what Hive can't do!)                 │
│  Row key: NEVER monotonic (hotspot!), use salt/reverse timestamp │
│  Write path: WAL → MemStore → HFile (flush on 128 MB)            │
│                                                                  │
│  SQOOP:                                                          │
│  Incremental import: --incremental append/lastmodified           │
│  --split-by: parallelize by column (default: primary key)        │
│  Export: NOT atomic → use staging table!                         │
│                                                                  │
│  OOZIE:                                                          │
│  Workflow: DAG of actions (run once)                             │
│  Coordinator: time + data trigger (production pipelines)         │
│                                                                  │
│  ZOOKEEPER:                                                      │
│  Leader election for NameNode/HMaster/YARN RM                    │
│  Always odd nodes (3/5/7) for quorum                             │
│  Ephemeral znodes = auto-deleted on crash = leader detection     │
│                                                                  │
│  TOP 5 THINGS TO SAY IN INTERVIEW:                               │
│  1. "External tables for all raw data — DROP TABLE is safe"      │
│  2. "MSCK REPAIR TABLE when Sqoop adds files outside Hive"       │
│  3. "Tez + vectorization + ORC can cut I/O and CPU; benchmark"  │
│  4. "HBase row key must not be monotonic — causes hotspot!"      │
│  5. "Oozie Coordinator = time + data trigger (smart scheduling)" │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

# Production Hadoop with Kerberos:
kinit -kt /etc/security/keytabs/hive.service.keytab hive/hiveserver2@ENTERPRISE.COM
# Login using keytab file (automated, no password prompt)
# Used in Oozie jobs, cron jobs, service accounts

klist
# Credentials cache: FILE:/tmp/krb5cc_1000
# Default principal: hive/hiveserver2@ENTERPRISE.COM
# Expires: Thu Jan 16 08:00:00 2025

# Renew ticket before it expires (important for long-running jobs!):
kinit -R
# ⚠️ If ticket expires mid-job → job fails with "Authentication failure"
# Solution: use long-lived keytabs + auto-renewal in production

RANGER POLICY EXAMPLES:

Policy 1: Hive table access
  Name: "Marketing team can read bookings"
  Resource: database=bookings_db, table=bookings, columns=*
  Permissions: user=marketing_role → SELECT
  Exclude: columns=credit_card_number (masked!)

Policy 2: Column masking (PII protection)
  Name: "Mask SSN for non-PII team"
  Resource: database=bookings_db, table=customers, column=ssn
  Masking: MASK (show only last 4 digits: ***-**-1234)
  Applied to: all users EXCEPT pii_approved_role

Policy 3: Row-level filter
  Name: "Regional teams see only their region"
  Resource: database=bookings_db, table=bookings
  Row filter: region = current_user_region()
  Applied to: regional_analyst role

RANGER AUDIT:
  Every access (success AND failure) is logged
  "Who accessed what, when, from where"
  Critical for compliance (GDPR, PCI-DSS, SOX)
  Stored in Solr or HDFS for long-term retention

<!-- Tune NameNode JVM heap for the live namespace and block maps held in RAM. -->
export HADOOP_NAMENODE_OPTS="-Xmx16g -Xms16g -XX:+UseG1GC"
<!-- -Xmx16g: max heap 16 GB -->
<!-- -Xms16g: initial heap = max heap (prevents resize pauses) -->
<!-- -XX:+UseG1GC: G1 garbage collector (low-pause, good for large heaps) -->
<!-- Default: only 1 GB → too small for clusters with >10 million files! -->

<!-- Rule for NameNode heap sizing: -->
<!-- 1 GB per 1 million files/blocks (rough estimate) -->
<!-- 10M files → 10 GB NameNode heap minimum -->
<!-- 100M files → 100 GB NameNode heap (need large server!) -->

<!-- DataNode JVM: less critical (doesn't keep metadata) -->
export HADOOP_DATANODE_OPTS="-Xmx4g -XX:+UseG1GC"

<!-- YARN NodeManager JVM: -->
export YARN_NODEMANAGER_OPTS="-Xmx4g -XX:+UseG1GC"

<!-- yarn-site.xml — Complete memory configuration -->

<!-- Node-level memory available to YARN: -->
<property>
  <name>yarn.nodemanager.resource.memory-mb</name>
  <value>49152</value>   <!-- 48 GB on a 64 GB server -->
  <!-- Reserve: 8 GB for OS + 8 GB for other services (HBase, DN, NM JVMs) -->
</property>

<!-- Node-level CPU vcores for YARN: -->
<property>
  <name>yarn.nodemanager.resource.cpu-vcores</name>
  <value>16</value>   <!-- All cores on a 16-core server -->
</property>

<!-- Scheduler min/max container sizes: -->
<property>
  <name>yarn.scheduler.minimum-allocation-mb</name>
  <value>1024</value>   <!-- No container gets < 1 GB -->
</property>
<property>
  <name>yarn.scheduler.maximum-allocation-mb</name>
  <value>16384</value>   <!-- No container gets > 16 GB -->
</property>

<!-- Memory increment: allocations happen in multiples of this: -->
<property>
  <name>yarn.scheduler.increment-allocation-mb</name>
  <value>512</value>
</property>

<!-- hdfs-site.xml performance settings -->

<!-- Block size: increase for large sequential files -->
<property>
  <name>dfs.blocksize</name>
  <value>268435456</value>   <!-- 256 MB for large files (default: 128 MB) -->
</property>

<!-- DataNode transfer thread pool (how many concurrent block transfers): -->
<property>
  <name>dfs.datanode.max.transfer.threads</name>
  <value>4096</value>   <!-- Default: 4096, can increase for high-throughput nodes -->
</property>

<!-- NameNode handler count (parallel RPC threads for client requests): -->
<property>
  <name>dfs.namenode.handler.count</name>
  <value>100</value>
  <!-- Default: 10 → way too small for large clusters! -->
  <!-- Rule: 20 * log2(cluster_size) -->
  <!-- 100 nodes → 20 * log2(100) = 20 * 6.6 = 132 → set to 100-150 -->
</property>

<!-- Short-circuit reads: client reads directly from local DataNode (no network!) -->
<property>
  <name>dfs.client.read.shortcircuit</name>
  <value>true</value>   <!-- If client is on same node as DataNode → read locally -->
</property>

<!-- mapred-site.xml performance settings — FULL REFERENCE -->

<!-- MAP task memory -->
<property>
  <name>mapreduce.map.memory.mb</name>
  <value>2048</value>
</property>
<property>
  <name>mapreduce.map.java.opts</name>
  <value>-Xmx1638m -XX:+UseG1GC</value>   <!-- 80% of container + G1GC -->
</property>

<!-- REDUCE task memory -->
<property>
  <name>mapreduce.reduce.memory.mb</name>
  <value>4096</value>
</property>
<property>
  <name>mapreduce.reduce.java.opts</name>
  <value>-Xmx3276m -XX:+UseG1GC</value>

<!-- Sort buffer: MOST IMPORTANT for shuffle performance -->
</property>
<property>
  <name>mapreduce.task.io.sort.mb</name>
  <value>512</value>   <!-- Default: 100 MB → increase to 512 MB! -->
</property>

<!-- Sort spill threshold: spill to disk when X% of sort buffer used -->
<property>
  <name>mapreduce.map.sort.spill.percent</name>
  <value>0.80</value>
</property>

<!-- Parallel copies during shuffle (how many parallel transfers from mappers): -->
<property>
  <name>mapreduce.reduce.shuffle.parallelcopies</name>
  <value>50</value>   <!-- Default: 5 → very low! Increase for faster shuffle -->
</property>

<!-- In-memory merge threshold (before writing merged spills to disk): -->
<property>
  <name>mapreduce.reduce.shuffle.input.buffer.percent</name>
  <value>0.70</value>   <!-- 70% of reducer heap used for in-memory shuffled data -->
</property>

<!-- Output compression -->
<property>
  <name>mapreduce.map.output.compress</name>
  <value>true</value>
</property>
<property>
  <name>mapreduce.map.output.compress.codec</name>
  <value>org.apache.hadoop.io.compress.SnappyCodec</value>
</property>

<!-- JVM reuse for jobs with many small tasks: -->
<property>
  <name>mapreduce.job.jvm.numtasks</name>
  <value>10</value>   <!-- Reuse same JVM for 10 tasks before restart -->
</property>

<!-- Number of parallel reducers: -->
<property>
  <name>mapreduce.job.reduces</name>
  <value>10</value>   <!-- Default: 1 — always set this based on data volume! -->
</property>

-- HIVE PERFORMANCE SETTINGS — run at start of session or in hive-site.xml

-- MUST-SET EVERY TIME:
SET hive.execution.engine=tez;                        -- NOT mr!
SET hive.vectorized.execution.enabled=true;           -- batch row processing
SET hive.vectorized.execution.reduce.enabled=true;
SET hive.cbo.enable=true;                             -- Cost-Based Optimizer
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;

-- JOIN OPTIMIZATION:
SET hive.auto.convert.join=true;                      -- auto map join for small tables
SET hive.mapjoin.smalltable.filesize=25000000;        -- 25 MB threshold
SET hive.auto.convert.join.noconditionaltask=true;
SET hive.auto.convert.join.noconditionaltask.size=20971520;  -- 20 MB

-- SKEW HANDLING:
SET hive.groupby.skewindata=true;                     -- 2-phase GROUP BY
SET hive.optimize.skewjoin=true;
SET hive.skewjoin.key=100000;

-- DYNAMIC PARTITIONING:
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.exec.max.dynamic.partitions=2000;
SET hive.exec.max.dynamic.partitions.pernode=500;

-- SMALL FILE MERGING (prevents small files after INSERT):
SET hive.merge.mapfiles=true;
SET hive.merge.mapredfiles=true;
SET hive.merge.smallfiles.avgsize=134217728;          -- target 128 MB
SET hive.merge.size.per.task=268435456;               -- max 256 MB per output file

-- PARALLELISM:
SET hive.exec.parallel=true;                          -- run independent stages in parallel
SET hive.exec.parallel.thread.number=8;               -- max parallel threads

-- LLAP (if available):
SET hive.llap.execution.mode=auto;

# HDFS Balancer: rebalance blocks across DataNodes
# When to run:
#   1. After adding new DataNodes (new nodes start empty!)
#   2. After replacing failed DataNodes
#   3. When some DataNodes are much more full than others

# Run balancer with specific bandwidth limit (don't saturate network):
hdfs balancer -threshold 10 -bandwidth 100m
# -threshold 10: a node is "balanced" if within 10% of average utilization
# -bandwidth 100m: use max 100 MB/s for rebalancing (don't flood production network)
# Without bandwidth limit: balancer uses ALL available bandwidth → impacts jobs!

# Check balance:
hdfs dfsadmin -report
# Look for: "DFS Used%" column across DataNodes
# If one node is at 90% and another at 20% → run balancer

# Intra-DataNode balancer (Hadoop 3 — balance disks within one DataNode):
hdfs diskbalancer -plan datanode-hostname
hdfs diskbalancer -execute plan.json
# Useful when one disk on a DataNode is much fuller than others

DATA SKEW = uneven data distribution across nodes/reducers

SYMPTOMS:
  - MapReduce: 99 reducers done, 1 reducer running for hours
  - Hive: GROUP BY query takes much longer than expected
  - YARN: one container using 10x memory of others

ROOT CAUSES:
  - Hot keys: "US" has 10M rows, "Fiji" has 100 rows → one reducer handles all US
  - Non-uniform partitioning: HashPartitioner with non-uniform key distribution
  - NULL values: all NULL keys go to same reducer by default

SOLUTIONS BY TYPE:

1. MAPREDUCE SKEW → Custom Partitioner + Salting
   Hot key "US" → spread as "US_0", "US_1", ..., "US_99" across 100 reducers
   Then second step: combine partial aggregations back

2. HIVE GROUP BY SKEW → skewindata
   SET hive.groupby.skewindata=true;
   Hive automatically uses 2-phase aggregation

3. HIVE JOIN SKEW → skewjoin
   SET hive.optimize.skewjoin=true;
   SET hive.skewjoin.key=100000;
   Hive splits join: hot keys via separate broadcast, rest via shuffle join

4. NULL KEY SKEW → handle NULLs explicitly
   -- Instead of: SELECT a.*, b.* FROM a JOIN b ON a.key = b.key
   -- (all NULLs in a.key go to one reducer!)
   -- Use:
   SELECT a.*, b.*
   FROM a JOIN b ON COALESCE(a.key, CONCAT('null_', RAND())) = b.key;
   -- Random null values → distributed across reducers

5. SPARK (if migrated): repartition(), broadcast(), salting still apply

WHAT: Move data from HDFS to cloud object storage (ADLS Gen2, S3, GCS)
      Keep same processing (Hive/MapReduce) but on cloud compute

HOW:
  Tool: DistCp (Distributed Copy — uses MapReduce to copy files in parallel)

hdfs distcp \
  hdfs://namenode:8020/user/hive/warehouse/ \
  wasbs://container@storageaccount.blob.core.windows.net/hive/warehouse/
  # wasbs = Windows Azure Storage Blob Secure

  Or to S3:
  hdfs distcp hdfs://namenode:8020/data/ s3a://my-bucket/data/

  For large datasets — incremental copy:
  hdfs distcp -update -skipcrccheck \
    hdfs://namenode/data/ \
    s3a://bucket/data/
  # -update: skip files already copied (incremental)
  # -skipcrccheck: S3 CRC differs from HDFS CRC (avoid false mismatches)

PROS:
  ✓ Minimal code changes (Hive still works with ADLS/S3 as backend)
  ✓ Low risk, fast migration
  ✓ Can keep Hive metastore pointing to new paths

CONS:
  ✗ Still paying for compute clusters (no serverless)
  ✗ MapReduce still disk-based and slow
  ✗ Misses opportunity to modernize processing

WHEN TO CHOOSE:
  → Budget limited, need quick win
  → Large Hadoop investment, can't rewrite apps
  → As Phase 1 of a larger migration

WHAT: Keep data in cloud storage, replace MapReduce processing with Spark
      Data: HDFS → S3/ADLS
      Processing: MapReduce Java → Spark (Python/Scala)

CHANGES:
  Old: MapReduce Job → YARN → HDFS
  New: Spark Job → Databricks/EMR/HDInsight → S3/ADLS

MIGRATION STEPS:
  1. Copy HDFS data to S3/ADLS (DistCp)
  2. Rewrite MapReduce jobs as Spark jobs (logic same, API different)
  3. Migrate Hive tables: update LOCATION to cloud paths
  4. Migrate Oozie workflows to Airflow/Databricks Workflows
  5. Migrate Sqoop imports to ADF/Glue/Spark JDBC

HIVE → SPARK SQL MIGRATION:
  Most Hive SQL is compatible with Spark SQL!
  -- Old Hive:
  SET hive.execution.engine=tez;
  INSERT INTO bookings_silver PARTITION (booking_date)
  SELECT ... FROM bookings_bronze WHERE booking_date >= '2024-01-01';

  -- New Spark SQL (almost identical!):
  spark.sql("""
    INSERT INTO bookings_silver PARTITION (booking_date)
    SELECT ... FROM bookings_bronze WHERE booking_date >= '2024-01-01'
  """)

PROS:
  ✓ Often faster for iterative or multi-stage work; validate Spark vs MapReduce on representative workloads
  ✓ Python/SQL (no more Java)
  ✓ Unified streaming + batch
  ✓ Lower compute costs (Spark finishes faster = fewer cluster-hours)

CONS:
  ✗ Rewriting all jobs takes time
  ✗ Spark expertise needed
  ✗ Still no ACID transactions without Delta Lake

WHAT: Full modernization to cloud-native lakehouse
  Storage: HDFS → ADLS Gen2 or S3
  Format:  ORC/Parquet → Delta Lake (ACID transactions!)
  Processing: Hive+Tez → Spark+Databricks
  Orchestration: Oozie → Databricks Workflows / ADF / Airflow
  Security: Ranger → Unity Catalog (Databricks)

HIVE TABLE → DELTA LAKE TABLE:
  Old Hive (ORC):
  CREATE EXTERNAL TABLE bookings (...)
  STORED AS ORC
  LOCATION 'hdfs://namenode/user/hive/warehouse/bookings/';

  New Delta Lake:
  CREATE TABLE bookings (...)
  USING DELTA
  LOCATION 'abfss://container@storage.dfs.core.windows.net/bookings/'

  Convert existing ORC to Delta (in Databricks):
  CONVERT TO DELTA parquet.`/mnt/data/bookings/`   -- if Parquet
  -- Or write new Delta table from Hive:
  spark.table("hive_db.bookings").write.format("delta").save("/mnt/delta/bookings")

MIGRATION TOOLING EXAMPLE: Databricks Lakebridge (verify current support and scope):
  - Official Databricks tool for Hive → Delta Lake migration
  - Auto-converts Hive DDL to Delta Lake CREATE TABLE statements
  - Migrates Hive Metastore → Unity Catalog
  - Translates HiveQL → Databricks SQL
  - Handles data copy from HDFS to cloud storage
  - Can assess compatibility: show what will break vs work automatically

PROS:
  ✓ Full ACID (what Hive ACID never quite delivered)
  ✓ Time travel / data versioning
  ✓ Schema enforcement + evolution
  ✓ Unified batch + streaming on same tables
  ✓ Unity Catalog = better governance than Ranger + Atlas
  ✓ Serverless (no cluster management)

CONS:
  ✗ Largest investment (rewrite + retrain team)
  ✗ Vendor lock-in (Databricks)
  ✗ Cost of migration project

WHEN TO CHOOSE:
  → Greenfield or major modernization initiative
  → Current Hadoop cost is very high (license + hardware)
  → Team is growing and needs SQL-first tooling
  → Need ACID and time travel on large tables

# Set EC policy on a directory:
hdfs ec -setPolicy -policy RS-6-3-1024k -path /data/archive/2020/
# All new files in /data/archive/2020/ will use EC instead of 3x replication

# Check current EC policy:
hdfs ec -getPolicy -path /data/archive/2020/

# List available EC policies:
hdfs ec -listPolicies

# Disable EC (back to replication) on a path:
hdfs ec -unsetPolicy -path /data/archive/2020/

"When I run SELECT country, COUNT(*) FROM bookings GROUP BY country in Hive:

1. Driver receives query, initiates session
2. Compiler sends query to Metastore: 'where is bookings table?'
   Metastore returns: HDFS path, ORC format, partition scheme
3. Parser builds AST (Abstract Syntax Tree)
4. Semantic analyzer validates: table exists? columns exist? permissions OK?
5. Optimizer (CBO if stats exist): decides join type, partition pruning, predicate pushdown
6. Physical plan: break into stages (Stage 1: MAP + partial agg, Stage 2: reduce+sort)
7. With Tez: translate plan to Tez DAG (not MapReduce chain)
8. Submit to YARN: YARN allocates containers for Tez workers
9. Each Tez task reads ORC file from HDFS, uses vectorization (1024 rows at a time)
10. Results aggregated, written to temp HDFS path
11. Driver returns results to client

Key optimizations I'd verify: execution.engine=tez, vectorization=true,
ORC format, partition pruning active, CBO stats current."

╔══════════════════════════════════════════════════════════════════════╗
║                  HADOOP COMPLETE CHEAT SHEET                          ║
║                    3-Day Interview Prep                                ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  ■ DAY 1: HDFS + YARN + MAPREDUCE                                    ║
║                                                                      ║
║  HDFS KEY NUMBERS: 128 MB blocks, 3x replication, 3s heartbeat        ║
║  NameNode: live namespace/block maps in RAM; persistent image+edits ║
║  DataNode: actual blocks, heartbeat every 3s, block report every 6h  ║
║                                                                      ║
║  NameNode HA = "JZ-FENCE":                                           ║
║    JournalNodes (quorum log) + ZooKeeper (election) + FENCING        ║
║    Fencing = kill old Active BEFORE promoting Standby (split-brain!) ║
║                                                                      ║
║  YARN = RM + NM + AM:                                                ║
║    ResourceManager: cluster boss (Scheduler + AppManager)            ║
║    NodeManager: per-node (runs containers, monitors resources)       ║
║    ApplicationMaster: per-JOB (negotiates resources, handles fails)  ║
║  Schedulers: FIFO (dev) / Capacity (multi-tenant) / Fair (mixed)     ║
║                                                                      ║
║  MapReduce = "I-Map-CBS-Reduce-O":                                   ║
║    Input → Map → Combiner → Buffer+Sort → Shuffle → Reduce → Output  ║
║    Shuffle = BOTTLENECK (disk + network + sort)                      ║
║    Combiner: cut shuffle 60-80% (only commutative+associative!)       ║
║    Speculative: duplicate slow tasks (DISABLE if side effects!)       ║
║                                                                      ║
║  Optimizations = "MCJ-COMP":                                         ║
║    Memory: sort buffer 100 MB→512 MB (biggest single win)              ║
║    Combiner: pre-aggregate locally                                    ║
║    JVM Reuse: jvm.numtasks=10 for many small tasks                   ║
║    Compression: Snappy shuffle, GZIP output                          ║
║    Output: LZO if output is input to next MR job (splittable!)       ║
║    More Reducers: nodes×containers×0.95 (not 1!)                    ║
║    Partitioner: custom for skewed keys                               ║
║                                                                      ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  ■ DAY 2: HIVE + ECOSYSTEM                                           ║
║                                                                      ║
║  Hive Architecture = "DMCTE":                                        ║
║    Driver → Metastore → Compiler → Tez Engine → Execution            ║
║    Metastore = MySQL backend, stores schema+HDFS paths               ║
║    ⚠️ Metastore DOWN → ALL Hive queries FAIL                         ║
║                                                                      ║
║  Internal vs External:                                               ║
║    Internal: DROP TABLE = data DELETED from HDFS!                    ║
║    External: DROP TABLE = metadata only, data SAFE                   ║
║    Usually External for raw/shared data when Hive must not own it    ║
║                                                                      ║
║  Partitioning:                                                       ║
║    Static: PARTITION (date='..') — one partition at a time           ║
║    Dynamic: nonstrict mode, Hive reads value from data               ║
║    MSCK REPAIR TABLE → sync new partitions from HDFS to Metastore    ║
║    Pruning: filter DIRECTLY on partition column (not YEAR(date)!)    ║
║                                                                      ║
║  Hive Optimization = "VECTOR-TOP":                                   ║
║    Engine=Tez (5-10x vs MapReduce), Vectorization (1024-row batches) ║
║    CBO + ANALYZE TABLE, Map Join (< 25 MB → broadcast)                ║
║    ORC format + predicate pushdown (skip stripes by min/max)         ║
║    skewindata=true (2-phase GROUP BY for hot keys)                   ║
║    SORT BY not ORDER BY (unless global order needed)                 ║
║    Merge small files: hive.merge.mapredfiles=true                    ║
║                                                                      ║
║  File Formats = "OPTA":                                              ║
║    ORC: Hive-native, ACID, predicate pushdown (best for Hive)        ║
║    Parquet: cross-tool Spark+Impala+Hive (best for multi-tool)       ║
║    Text: can compress; row-oriented, no column pruning                ║
║          use for landing/interchange; check codec splittability       ║
║    Avro: schema evolution, Kafka/Sqoop landing (row-based)           ║
║                                                                      ║
║  HBase: NoSQL random R/W on HDFS                                     ║
║    Row key: NEVER monotonic (hotspot!) → use salt/reverse timestamp  ║
║    Write: WAL → MemStore → HFile flush at 128 MB                      ║
║                                                                      ║
║  Sqoop: RDBMS ↔ HDFS                                                 ║
║    Incremental: --incremental append (new rows) or lastmodified      ║
║    --num-mappers: parallel DB connections (4-8 max!)                 ║
║    Export NOT atomic → use staging table!                            ║
║                                                                      ║
║  Oozie: Workflow (run once) + Coordinator (time+data trigger)        ║
║  ZooKeeper: leader election (NameNode/HMaster/YARN RM)               ║
║    Always odd nodes (3/5/7), ephemeral znodes for leader detection   ║
║  Flume: Source → Channel → Sink for log ingestion                    ║
║    File channel = durable; Memory channel = fast but lossy           ║
║                                                                      ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  ■ DAY 3: PERFORMANCE, SECURITY, MIGRATION                           ║
║                                                                      ║
║  Security = "KRK":                                                   ║
║    Kerberos: Authentication (kinit → TGT → Service Ticket)           ║
║    Ranger: Authorization (table/column/row-level policies + audit)   ║
║    Knox: Gateway (single HTTPS entry + SSL + SSO)                    ║
║                                                                      ║
║  JVM Tuning:                                                         ║
║    NameNode: -Xmx32g -Xms32g -XX:+UseG1GC                           ║
║    NameNode heap: 1 GB per 1 million files (150 bytes/file)           ║
║    G1GC: short predictable pauses vs CMS long stop-the-world         ║
║                                                                      ║
║  MapReduce Critical Settings:                                        ║
║    mapreduce.task.io.sort.mb=512     (default 100 = too small!)      ║
║    mapreduce.map.output.compress=true + Snappy codec                 ║
║    mapreduce.reduce.shuffle.parallelcopies=50 (default 5)            ║
║    mapreduce.job.jvm.numtasks=10 (JVM reuse for small task jobs)     ║
║    mapreduce.job.reduces=N (never leave at default 1!)               ║
║                                                                      ║
║  Data Skew Solutions:                                                ║
║    MapReduce: salting (US → US_0...US_99) + custom Partitioner       ║
║    Hive GROUP BY: hive.groupby.skewindata=true (2-phase agg)         ║
║    Hive JOIN: hive.optimize.skewjoin=true                            ║
║    NULL keys: COALESCE(key, CONCAT('null_', RAND()))                 ║
║                                                                      ║
║  Erasure Coding (Hadoop 3):                                          ║
║    RS-6-3: 6 data + 3 parity blocks = 1.5x overhead vs 3x           ║
║    50% storage savings for cold/archive data                         ║
║    ⚠️ High CPU for reconstruction → cold data only!                  ║
║                                                                      ║
║  Cloud Migration = "LRR":                                            ║
║    Lift-and-Shift: DistCp HDFS→S3/ADLS, same code                   ║
║    Replatform: MapReduce → Spark (benchmark representative jobs)     ║
║    Refactor: Hive → Delta Lake; evaluate current migration tooling   ║
║    Always: parallel run → compare → cutover (never big-bang!)        ║
║                                                                      ║
║  HADOOP vs SPARK:                                                    ║
║    MapReduce: disk-based, Java, batch, slow (minutes to hours)       ║
║    Spark: in-memory, Python/SQL, streaming+batch (seconds-minutes)   ║
║    New work: choose by requirements; MapReduce may remain in estates ║
║                                                                      ║
╠══════════════════════════════════════════════════════════════════════╣
║                                                                      ║
║  TOP 10 THINGS TO SAY IN INTERVIEW:                                  ║
║                                                                      ║
║  1. "NameNode HA: JournalNodes + ZooKeeper + FENCING (splits-brain)" ║
║  2. "Shuffle is bottleneck: sort buffer 100 MB → 512 MB is biggest win"║
║  3. "External tables for all raw/shared data — DROP TABLE is safe"   ║
║  4. "MSCK REPAIR TABLE when Sqoop adds partitions outside Hive"      ║
║  5. "Tez + vectorization + ORC can cut I/O and CPU; benchmark"      ║
║  6. "HBase row key: NEVER monotonic (hotspot) → use salting"         ║
║  7. "Kerberos=authentication, Ranger=authorization, Knox=gateway"    ║
║  8. "Erasure Coding saves 50% storage — use for cold/archive data"   ║
║  9. "Migration: Lift-Shift→ Replatform→ Refactor (never big-bang)"  ║
║  10. "skewindata=true for GROUP BY skew, custom partitioner for MR"  ║
║                                                                      ║
║  10-YEAR ENGINEER FRAMING:                                           ║
║  "I've not just run Hadoop queries — I've tuned NameNode heap,       ║
║   fixed GC pauses causing DataNode timeouts, debugged ZKFC failover, ║
║   implemented salting for skewed keys, and migrated pipelines from   ║
║   MapReduce to Spark and then to Databricks lakehouses."             ║
║                                                                      ║
╚══════════════════════════════════════════════════════════════════════╝

hdfs dfs -ls <path>
hdfs dfs -ls -R <path>          # Recursive listing (all subdirectories)
hdfs dfs -ls -h <path>          # Human-readable file sizes (KB, MB, GB)

# List all files in the bookings directory
hdfs dfs -ls /data/travelco/bookings/
# Output:
# -rw-r--r--   3 krishna hadoop  1073741824 2026-03-25 14:30 /data/travelco/bookings/booking_2026.orc

# Recursive list to see all partitions under a Hive table
hdfs dfs -ls -R /user/hive/warehouse/bookings_db.db/flights/

# List with human-readable sizes
hdfs dfs -ls -h /data/travelco/bookings/
# Output: shows 1.0 G instead of 1073741824

hdfs dfs -mkdir <path>
hdfs dfs -mkdir -p <path>       # Create parents (like Linux mkdir -p)

# Create a single directory (parent must exist)
hdfs dfs -mkdir /data/travelco/

# Create full directory tree in one shot
hdfs dfs -mkdir -p /data/travelco/bookings/year=2026/month=03/day=25

# Common pattern: create staging + final directories
hdfs dfs -mkdir -p /data/staging/flights/
hdfs dfs -mkdir -p /data/processed/flights/

hdfs dfs -put <localPath> <hdfsPath>
hdfs dfs -copyFromLocal <localPath> <hdfsPath>
hdfs dfs -put -f <localPath> <hdfsPath>     # Overwrite if exists

# Upload a local CSV to HDFS
hdfs dfs -put /home/krishna/booking_data.csv /data/travelco/staging/

# Upload and overwrite if file already exists
hdfs dfs -put -f /home/krishna/daily_export.csv /data/travelco/staging/daily_export.csv

# Upload multiple files at once
hdfs dfs -put /home/krishna/logs/*.log /data/travelco/raw_logs/

hdfs dfs -get <hdfsPath> <localPath>
hdfs dfs -copyToLocal <hdfsPath> <localPath>

# Download a single file from HDFS
hdfs dfs -get /data/travelco/reports/monthly_summary.csv /home/krishna/

# Download an entire directory
hdfs dfs -get /data/travelco/bookings/year=2026/month=03/ /home/krishna/march_data/

# Download and overwrite existing local file
hdfs dfs -get -f /data/travelco/reports/latest.csv /home/krishna/latest.csv

hdfs dfs -cat <hdfsPath>
hdfs dfs -head <hdfsPath>           # First 1 KB of file
hdfs dfs -tail <hdfsPath>           # Last 1 KB of file
hdfs dfs -cat <hdfsPath> | head -20 # First 20 lines (pipe to local head)

# Quick peek at a small file
hdfs dfs -cat /data/travelco/config/etl_params.json

# View first 1 KB of a large log file
hdfs dfs -head /data/travelco/raw_logs/access.log

# View last 1 KB (check latest entries)
hdfs dfs -tail /data/travelco/raw_logs/access.log

# Practical: view first 20 lines of a CSV to check schema
hdfs dfs -cat /data/travelco/staging/bookings.csv | head -20

hdfs dfs -mv <source> <destination>
hdfs dfs -cp <source> <destination>

# Move processed files from staging to final location
hdfs dfs -mv /data/staging/bookings_2026.orc /data/processed/bookings_2026.orc

# Rename a file
hdfs dfs -mv /data/travelco/old_name.csv /data/travelco/new_name.csv

# Copy a file (creates a new copy with full replication)
hdfs dfs -cp /data/travelco/bookings/current.orc /data/travelco/bookings/backup_current.orc

# Move entire directory
hdfs dfs -mv /data/staging/batch_20260325/ /data/processed/batch_20260325/

hdfs dfs -rm <filePath>                 # Delete a file
hdfs dfs -rm -r <directoryPath>         # Delete directory recursively
hdfs dfs -rm -r -skipTrash <path>       # Permanent delete (bypass Trash)
hdfs dfs -rm -r -f <path>              # Force delete (no error if path missing)

# Delete a single file
hdfs dfs -rm /data/staging/temp_file.csv

# Delete a directory and all contents
hdfs dfs -rm -r /data/staging/batch_20260324/

# Permanent delete when disk is full (bypass trash)
hdfs dfs -rm -r -skipTrash /data/old_logs/2024/

# Safe delete in scripts (no error if already deleted)
hdfs dfs -rm -r -f /data/staging/temp_dir/

hdfs dfs -du <path>                  # Size of each item in the path
hdfs dfs -du -s <path>              # Summary (total size of directory)
hdfs dfs -du -s -h <path>           # Human-readable summary

# Check size of each subdirectory
hdfs dfs -du -h /data/travelco/
# Output:
# 12.5 G  37.4 G  /data/travelco/bookings
# 3.2 G   9.6 G   /data/travelco/flights
# 156.7 M 470.1 M /data/travelco/config
# First number = raw size, Second number = space with replication

# Total size of one directory
hdfs dfs -du -s -h /data/travelco/bookings/
# Output: 12.5 G  37.4 G  /data/travelco/bookings

# Check which partitions are largest (find data skew)
hdfs dfs -du -h /user/hive/warehouse/bookings_db.db/flights/year=2026/

hdfs dfs -chmod <permissions> <path>
hdfs dfs -chmod -R <permissions> <path>     # Recursive
hdfs dfs -chown <owner>:<group> <path>
hdfs dfs -chown -R <owner>:<group> <path>   # Recursive

# Give read-write-execute to owner, read-execute to group and others
hdfs dfs -chmod 755 /data/travelco/bookings/

# Recursively set permissions for a Hive table directory
hdfs dfs -chmod -R 770 /user/hive/warehouse/bookings_db.db/

# Change ownership (for service accounts)
hdfs dfs -chown -R hive:hadoop /user/hive/warehouse/bookings_db.db/

# Give only the ETL service account write access
hdfs dfs -chown etl_user:etl_group /data/staging/
hdfs dfs -chmod 750 /data/staging/

hdfs dfs -count <path>
hdfs dfs -count -q <path>          # Include quota information
hdfs dfs -count -h <path>          # Human-readable sizes

# Count files under a Hive table directory
hdfs dfs -count /user/hive/warehouse/bookings_db.db/flights/
# Output: 365  4380  45231965798  /user/hive/warehouse/bookings_db.db/flights/
# Meaning: 365 directories, 4380 files, 45.2 GB total

# Human-readable format
hdfs dfs -count -h /user/hive/warehouse/bookings_db.db/flights/
# Output: 365  4380  42.1 G  /user/hive/warehouse/bookings_db.db/flights/

# Check quota usage
hdfs dfs -count -q -h /data/travelco/

hdfs dfs -stat <format> <path>
# Format specifiers:
# %b = file size (bytes), %n = name, %o = block size
# %r = replication factor, %y = modification date

# Get replication factor of a file
hdfs dfs -stat %r /data/travelco/bookings/booking_2026.orc
# Output: 3

# Get block size
hdfs dfs -stat %o /data/travelco/bookings/booking_2026.orc
# Output: 134217728  (128 MB)

# Get file size and modification time
hdfs dfs -stat "%b %y" /data/travelco/bookings/booking_2026.orc
# Output: 1073741824 2026-03-25 14:30:22

# Quick check: is replication factor correct?
hdfs dfs -stat "%n: replication=%r, blocksize=%o" /data/travelco/bookings/booking_2026.orc
# Output: booking_2026.orc: replication=3, blocksize=134217728

hdfs dfs -touchz <path>

# Create a _SUCCESS flag file to signal ETL completion
hdfs dfs -touchz /data/travelco/bookings/year=2026/month=03/day=25/_SUCCESS

# Create a lock file for job coordination
hdfs dfs -touchz /data/travelco/locks/daily_etl.lock

# Create marker file for downstream pipelines
hdfs dfs -touchz /data/staging/batch_complete_20260325

hdfs dfs -setrep <replication> <path>
hdfs dfs -setrep -R <replication> <path>    # Recursive
hdfs dfs -setrep -w <replication> <path>    # Wait until replication completes

# Increase replication for critical booking data
hdfs dfs -setrep 5 /data/travelco/bookings/current_month.orc

# Reduce replication for cold/archive data to save space
hdfs dfs -setrep -R 2 /data/travelco/archive/2024/

# Set replication for hot data and wait for completion
hdfs dfs -setrep -w 3 /data/travelco/bookings/today.orc

# Reduce replication on staging data (temporary, don't need 3 copies)
hdfs dfs -setrep -R 1 /data/staging/

hdfs dfs -getmerge <hdfsDir> <localFile>
hdfs dfs -getmerge -nl <hdfsDir> <localFile>   # Add newline between files

# Merge all MapReduce output parts into one file
hdfs dfs -getmerge /output/job_20260325/ /home/krishna/merged_output.csv

# Merge Hive query output (multiple part-00000 files)
hdfs dfs -getmerge /user/hive/warehouse/temp_results/ /home/krishna/query_results.txt

# Add newline separator between merged files
hdfs dfs -getmerge -nl /output/daily_reports/ /home/krishna/full_report.csv

hdfs fsck <path> [options]
hdfs fsck <path> -files              # List all files
hdfs fsck <path> -blocks             # Show block information
hdfs fsck <path> -locations          # Show block locations (which DataNodes)
hdfs fsck <path> -racks              # Show rack information
hdfs fsck <path> -files -blocks -locations   # Full detail

# Check overall cluster health
hdfs fsck /

# Check health of a specific directory
hdfs fsck /data/travelco/bookings/ -files -blocks -locations
# Output shows:
# Total files: 4380
# Total blocks: 13140
# Minimally replicated blocks: 13140 (100.0%)
# Under-replicated blocks: 0 (0.0%)
# Missing blocks: 0 (0.0%)
# Corrupt blocks: 0

# Find which DataNodes hold blocks of a specific file
hdfs fsck /data/travelco/bookings/booking_2026.orc -files -blocks -locations
# Output:
# /data/travelco/bookings/booking_2026.orc 1073741824 bytes, 8 block(s):
#   blk_1073741825 len=134217728 [10.0.0.5:9866, 10.0.0.7:9866, 10.0.0.12:9866]
#   blk_1073741826 len=134217728 [10.0.0.3:9866, 10.0.0.8:9866, 10.0.0.15:9866]

# Check for corrupt or missing blocks across the cluster
hdfs fsck / -list-corruptfileblocks

hdfs balancer
hdfs balancer -threshold <percentage>     # Default 10%
hdfs balancer -policy datanode            # Balance by DataNode usage

# Run balancer with default 10% threshold
# (stops when all nodes are within 10% of average utilization)
hdfs balancer

# Tighter balance — within 5% of average
hdfs balancer -threshold 5

# Run with bandwidth limit to avoid impacting production jobs
# (set in hdfs-site.xml: dfs.datanode.balance.bandwidthPerSec)
hdfs balancer -threshold 10

# Check balance status — look at utilization spread
hdfs dfsadmin -report | grep "DFS Used%"

hdfs dfsadmin -report
hdfs dfsadmin -report -live           # Only live DataNodes
hdfs dfsadmin -report -dead           # Only dead DataNodes

# Full cluster health report
hdfs dfsadmin -report
# Output:
# Configured Capacity: 500 TB
# Present Capacity: 480 TB
# DFS Remaining: 200 TB (41.67%)
# DFS Used: 280 TB (58.33%)
# Under replicated blocks: 0
# Blocks with corrupt replicas: 0
# Missing blocks: 0
#
# ----- Live DataNodes (50) -----
# Name: 10.0.0.5:9866 (datanode05.cluster.local)
# Decommission Status: Normal
# Configured Capacity: 10 TB
# DFS Used: 5.6 TB (56.00%)
# ...

# Quick check: are any DataNodes dead?
hdfs dfsadmin -report -dead

hdfs dfsadmin -safemode get            # Check if safe mode is ON/OFF
hdfs dfsadmin -safemode enter          # Manually enter safe mode
hdfs dfsadmin -safemode leave          # Manually leave safe mode
hdfs dfsadmin -safemode wait           # Block until safe mode exits

# Check safe mode status
hdfs dfsadmin -safemode get
# Output: Safe mode is OFF

# Enter safe mode before maintenance
hdfs dfsadmin -safemode enter
# Do maintenance (e.g., snapshot, config change)
hdfs dfsadmin -safemode leave

# In scripts: wait for safe mode to finish after cluster restart
hdfs dfsadmin -safemode wait
echo "HDFS is ready, starting ETL jobs..."

yarn application -list                               # Running apps
yarn application -list -appStates ALL                # All states
yarn application -list -appStates FINISHED           # Completed apps
yarn application -list -appStates FAILED,KILLED      # Failed or killed
yarn application -list -appTypes SPARK               # Only Spark apps

# See what's currently running on the cluster
yarn application -list
# Output:
# Application-Id          Name                  State    Queue     Progress
# application_1234_0001   daily_booking_etl     RUNNING  default   65%
# application_1234_0002   hive_query_xyz        RUNNING  adhoc     30%

# Check how many jobs finished today
yarn application -list -appStates FINISHED

# Find all failed Spark jobs
yarn application -list -appStates FAILED -appTypes SPARK

yarn application -status <applicationId>

# Check status of a specific application
yarn application -status application_1711350000000_0042
# Output:
# Application Report:
#   Application-Id: application_1711350000000_0042
#   Application-Name: daily_booking_etl
#   Application-Type: SPARK
#   State: RUNNING
#   Final-Status: UNDEFINED
#   Progress: 65%
#   Queue: production
#   AM Host: datanode15.cluster.local
#   Allocated Resources: <memory:40960 MB, vCores:16>
#   Running Containers: 8
#   Start-Time: 1711350120000

yarn application -kill <applicationId>

# Kill a stuck Hive query that's consuming all cluster resources
yarn application -kill application_1711350000000_0042
# Output: Killing application application_1711350000000_0042
# Application application_1711350000000_0042 has been killed.

# Common scenario: kill all stuck applications in a loop
for app_id in $(yarn application -list -appStates RUNNING | grep "stuck_job" | awk '{print $1}'); do
    yarn application -kill $app_id
done

yarn logs -applicationId <appId>
yarn logs -applicationId <appId> -containerId <containerId>   # Specific container
yarn logs -applicationId <appId> -nodeAddress <nodeAddress>    # Specific node
yarn logs -applicationId <appId> -log_files stderr             # Only stderr

# Get all logs for a failed application
yarn logs -applicationId application_1711350000000_0042 > /home/krishna/job_logs.txt

# Get only stderr (where exceptions appear)
yarn logs -applicationId application_1711350000000_0042 -log_files stderr

# Get logs from a specific container (useful for debugging specific task failures)
yarn logs -applicationId application_1711350000000_0042 \
  -containerId container_1711350000000_0042_01_000005

# Get logs from a specific node
yarn logs -applicationId application_1711350000000_0042 \
  -nodeAddress datanode15.cluster.local:8041

yarn node -list                    # All nodes
yarn node -list -states RUNNING    # Only active nodes
yarn node -list -all               # Include decommissioned/lost nodes

# List all active NodeManagers
yarn node -list -states RUNNING
# Output:
# Node-Id                  Node-State  Node-Http-Address     Containers
# datanode01:45454         RUNNING     datanode01:8042       4
# datanode02:45454         RUNNING     datanode02:8042       6
# datanode03:45454         RUNNING     datanode03:8042       3

# Check for unhealthy or lost nodes
yarn node -list -all

yarn queue -status <queueName>

# Check the production queue status
yarn queue -status production
# Output:
# Queue Name: production
# State: RUNNING
# Capacity: 60.0%
# Current Capacity: 45.0%
# Maximum Capacity: 80.0%
# Default Node Label: <DEFAULT>
# Number of Applications: 3

# Check the adhoc queue
yarn queue -status adhoc

yarn top

# Real-time resource monitoring
yarn top
# Output (refreshes every 3 seconds):
# JEEP CLUSTER SUMMARY: 50 NodeManagers
# Memory: 500 GB / 800 GB used (62.5%)
# VCores: 200 / 400 used (50.0%)
# Queue    Capacity  Used  Apps  Containers
# root.prod  60%     45%    3      24
# root.adhoc 30%     15%    2      8
# root.dev   10%     2%     1      2

beeline -u "jdbc:hive2://<host>:<port>/<database>"
beeline -u "jdbc:hive2://<host>:<port>/<database>" -n <username> -p <password>
beeline -u "jdbc:hive2://<host>:10000/default" --hiveconf hive.execution.engine=tez

# Connect to HiveServer2 on default port
beeline -u "jdbc:hive2://hiveserver.cluster.local:10000/default"

# Connect with Kerberos authentication
beeline -u "jdbc:hive2://hiveserver.cluster.local:10000/default;principal=hive/_HOST@REALM.COM"

# Connect and run a single query (non-interactive)
beeline -u "jdbc:hive2://hiveserver.cluster.local:10000/default" \
  -e "SELECT count(*) FROM bookings WHERE year=2026"

# Connect and run a script file
beeline -u "jdbc:hive2://hiveserver.cluster.local:10000/default" \
  -f /home/krishna/etl_daily.hql

SHOW DATABASES;
SHOW TABLES;
SHOW TABLES IN <database>;
DESCRIBE <table>;
DESCRIBE FORMATTED <table>;        -- Full metadata including location, format, partitions
DESCRIBE EXTENDED <table>;         -- Similar but less readable
SHOW PARTITIONS <table>;
SHOW CREATE TABLE <table>;         -- DDL to recreate the table

-- List all databases
SHOW DATABASES;

-- Switch to bookings database
USE bookings_db;

-- List all tables
SHOW TABLES;

-- Quick schema check
DESCRIBE flights;
-- Output:
-- flight_id    int
-- origin       string
-- destination  string
-- departure    timestamp
-- year         int        (partition column)
-- month        int        (partition column)

-- Full metadata (CRITICAL for interviews)
DESCRIBE FORMATTED flights;
-- Shows: location (HDFS path), InputFormat, OutputFormat, SerDe,
-- partition columns, table type (MANAGED/EXTERNAL), creation time

-- See all partitions
SHOW PARTITIONS flights;
-- Output:
-- year=2025/month=01
-- year=2025/month=02
-- ...
-- year=2026/month=03

-- Get exact DDL
SHOW CREATE TABLE flights;

-- INTERNAL (Managed) TABLE: Hive owns the data. DROP TABLE = data deleted.
CREATE TABLE bookings (
    booking_id   BIGINT,
    passenger    STRING,
    flight_code  STRING,
    amount       DOUBLE,
    booking_time TIMESTAMP
)
STORED AS ORC
TBLPROPERTIES ('orc.compress'='SNAPPY');

-- EXTERNAL TABLE: Hive only manages metadata. DROP TABLE = data stays.
CREATE EXTERNAL TABLE flights_raw (
    flight_id    INT,
    origin       STRING,
    destination  STRING,
    departure    STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION '/data/travelco/raw/flights/';

-- PARTITIONED TABLE (most common in production)
CREATE EXTERNAL TABLE bookings_partitioned (
    booking_id   BIGINT,
    passenger    STRING,
    flight_code  STRING,
    amount       DOUBLE
)
PARTITIONED BY (year INT, month INT, day INT)
STORED AS ORC
LOCATION '/data/travelco/bookings/';

-- BUCKETED TABLE (for optimized joins)
CREATE TABLE bookings_bucketed (
    booking_id   BIGINT,
    passenger    STRING,
    flight_code  STRING,
    amount       DOUBLE
)
CLUSTERED BY (booking_id) INTO 32 BUCKETS
STORED AS ORC;

-- LOAD DATA: moves a file into the table's HDFS directory (no transformation)
LOAD DATA INPATH '<hdfs_path>' INTO TABLE <table>;
LOAD DATA INPATH '<hdfs_path>' OVERWRITE INTO TABLE <table>;
LOAD DATA LOCAL INPATH '<local_path>' INTO TABLE <table>;

-- INSERT INTO: appends query results
INSERT INTO TABLE <target> SELECT * FROM <source>;

-- INSERT OVERWRITE: replaces all data (or partition)
INSERT OVERWRITE TABLE <target> SELECT * FROM <source>;

-- DYNAMIC PARTITION INSERT
INSERT OVERWRITE TABLE bookings PARTITION (year, month, day)
SELECT booking_id, passenger, flight_code, amount, year, month, day
FROM staging_bookings;

-- Load a CSV file from HDFS into a raw table
LOAD DATA INPATH '/data/staging/bookings_20260325.csv' INTO TABLE bookings_raw;

-- Load from local filesystem (copies to HDFS first)
LOAD DATA LOCAL INPATH '/home/krishna/test_data.csv' INTO TABLE test_table;

-- ETL pattern: transform and load into partitioned ORC table
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;

INSERT OVERWRITE TABLE bookings_orc PARTITION (year, month)
SELECT
    booking_id, passenger, flight_code, amount,
    year(booking_time) AS year,
    month(booking_time) AS month
FROM bookings_raw
WHERE booking_date = '2026-03-25';

-- Overwrite a specific partition only
INSERT OVERWRITE TABLE bookings_orc PARTITION (year=2026, month=3)
SELECT booking_id, passenger, flight_code, amount
FROM bookings_raw
WHERE year(booking_time) = 2026 AND month(booking_time) = 3;

-- Add a partition manually (static partitioning)
ALTER TABLE bookings ADD PARTITION (year=2026, month=3, day=25)
LOCATION '/data/travelco/bookings/year=2026/month=03/day=25';

-- Drop a partition (data deleted for managed tables, kept for external)
ALTER TABLE bookings DROP PARTITION (year=2024, month=1);

-- Rename table
ALTER TABLE old_bookings RENAME TO bookings_archive;

-- Add a column
ALTER TABLE bookings ADD COLUMNS (loyalty_tier STRING);

-- Change column name/type
ALTER TABLE bookings CHANGE old_column_name new_column_name BIGINT;

-- Change table properties
ALTER TABLE bookings SET TBLPROPERTIES ('orc.compress'='ZLIB');

-- Change HDFS location
ALTER TABLE bookings SET LOCATION '/data/travelco/bookings_v2/';

MSCK REPAIR TABLE <table>;

-- Scenario: Spark/ETL wrote new partition directories to HDFS
-- HDFS has: /data/bookings/year=2026/month=03/day=25/
-- But Hive doesn't know about it yet

-- Sync Hive Metastore with HDFS
MSCK REPAIR TABLE bookings;
-- Output: Partitions not in metastore: bookings:year=2026/month=03/day=25
-- Now the partition is queryable

-- Verify
SHOW PARTITIONS bookings;

EXPLAIN <query>;
EXPLAIN EXTENDED <query>;           -- More detail
EXPLAIN FORMATTED <query>;          -- Readable JSON format

-- See execution plan for a join query
EXPLAIN
SELECT b.booking_id, f.origin, f.destination
FROM bookings b
JOIN flights f ON b.flight_code = f.flight_code
WHERE b.year = 2026 AND b.month = 3;

-- Output shows:
-- Stage-1: Map (reads bookings, applies partition filter)
-- Stage-2: Map (reads flights)
-- Stage-3: Reduce (shuffle join on flight_code)

-- Check if partition pruning is working
EXPLAIN
SELECT count(*) FROM bookings WHERE year = 2026;
-- Look for: filterExpr: (year = 2026) — means partition pruning is active

-- Select Tez for DAG execution; verify engine availability and benchmark the workload
SET hive.execution.engine=tez;

-- Enable vectorized execution (process 1024 rows at a time instead of 1)
SET hive.vectorized.execution.enabled=true;
SET hive.vectorized.execution.reduce.enabled=true;

-- Enable dynamic partitioning
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;

-- Enable map join (broadcast small table)
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=25000000;  -- 25 MB threshold

-- Enable CBO (Cost-Based Optimizer)
SET hive.cbo.enable=true;
SET hive.compute.query.using.stats=true;
SET hive.stats.fetch.column.stats=true;

-- Enable compression
SET hive.exec.compress.output=true;
SET mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

-- Control parallelism
SET hive.exec.parallel=true;
SET hive.exec.parallel.thread.number=8;