1. Get ingredients from fridge (LOAD from HDFS)
2. Cut the vegetables (FILTER/FOREACH)
3. Mix everything together (JOIN/GROUP)
4. Cook it (execution on MapReduce)
5. Serve the dish (STORE back to HDFS)

SELECT product_id, SUM(revenue) as total
FROM sales
WHERE country = 'India'
GROUP BY product_id
ORDER BY total DESC
LIMIT 10;

 PIG HIVE
────────────────────────────────────────────────────────────
Language: Pig Latin (procedural) HiveQL (SQL-like)
Abstraction: DataFlow pipeline Relational tables
Best for: ETL, transformations Reporting, analytics
User type: Data Engineers Data Analysts / SQL users
Schema: Schema on load Schema in metastore
Execution: MapReduce / Tez / Spark MapReduce / Tez / Spark
Debugging: Grunt shell (DUMP) Beeline CLI (SELECT)
Flexibility: Very high (code logic) Medium (SQL constraints)

-- Load raw logs from HDFS
logs = LOAD '/logs/access.log' USING TextLoader() AS (line:chararray);

-- Parse each log line (mapper step essentially)
parsed = FOREACH logs GENERATE
 STRSPLIT(line, ' ') AS tokens;

-- Extract fields we care about
extracted = FOREACH parsed GENERATE
 tokens.$0 AS ip,
 tokens.$6 AS url,
 tokens.$8 AS status;

-- Filter only 500 errors
errors = FILTER extracted BY status == '500';

-- Count errors per IP address 
grouped = GROUP errors BY ip;
counts = FOREACH grouped GENERATE group AS ip, COUNT(errors) AS error_count;

-- Sort by count descending
sorted = ORDER counts BY error_count DESC;

-- Store results back to HDFS
STORE sorted INTO '/analytics/500_errors' USING PigStorage('|');

192.168.1.100|52
10.0.0.25|38
172.16.0.8|21
...

CREATE EXTERNAL TABLE error_counts (ip STRING, error_count INT)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '|'
LOCATION '/analytics/500_errors';

-- Now analysts can query with SQL!
SELECT ip, error_count
FROM error_counts
WHERE error_count > 20
ORDER BY error_count DESC;

-- Step 1: External table pointing to raw JSON events in HDFS
CREATE EXTERNAL TABLE events_raw (line STRING)
LOCATION '/data/events/raw/';

-- Step 2: Structured table partitioned by date
CREATE TABLE events (
 userid STRING,
 ts BIGINT,
 action STRING
)
PARTITIONED BY (dt STRING)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS ORC;

-- Step 3: Transform and load for a specific date
INSERT INTO events PARTITION(dt='20260226')
SELECT
 get_json_object(line, '$.user') AS userid,
 get_json_object(line, '$.timestamp') AS ts,
 get_json_object(line, '$.action') AS action
FROM events_raw
WHERE get_json_object(line, '$.dt') = '20260226';

-- Load CSV data
sales = LOAD '/data/sales.csv'
 USING PigStorage(',')
 AS (id:int, date:chararray, product:chararray, amount:float, country:chararray);

-- Load JSON data
events = LOAD '/data/events/'
 USING JsonLoader()
 AS (user:chararray, action:chararray, ts:long);

-- Load without schema (schema-on-read style)
raw_data = LOAD '/data/mystery.txt' AS (line:chararray);

-- Simple condition
big_sales = FILTER sales BY amount > 10000.0;

-- Multiple conditions
india_high_value = FILTER sales BY country == 'India' AND amount > 5000.0;

-- Regex matching
product_pages = FILTER events BY action MATCHES '/product/.*';

-- Null check
clean_data = FILTER sales BY date IS NOT NULL;

-- Select only specific columns (projection)
slim = FOREACH sales GENERATE id, amount;

-- Calculate a new column
with_tax = FOREACH sales GENERATE
 id,
 amount,
 amount * 0.18 AS tax,
 amount * 1.18 AS total_with_tax;

-- String manipulation
upper_country = FOREACH sales GENERATE
 id,
 UPPER(country) AS country_upper,
 SUBSTRING(date, 0, 4) AS year;

-- Using a UDF (User-Defined Function)
enriched = FOREACH sales GENERATE
 id,
 mypackage.MyUDF(product) AS category;

-- Group by one field
by_country = GROUP sales BY country;

-- Group by multiple fields
by_country_year = GROUP sales BY (country, SUBSTRING(date, 0, 4));

-- Group ALL into one bag (for grand totals)
all_together = GROUP sales ALL;

-- Then aggregate with FOREACH:
country_totals = FOREACH by_country GENERATE
 group AS country, -- the group key
 COUNT(sales) AS num_orders,
 SUM(sales.amount) AS total_revenue,
 AVG(sales.amount) AS avg_order_value,
 MAX(sales.amount) AS biggest_order;

customers = LOAD '/data/customers.csv' AS (cid:int, name:chararray, email:chararray);
orders = LOAD '/data/orders.csv' AS (oid:int, cid:int, amount:float);

-- Inner join: only customers WITH orders
cust_orders = JOIN customers BY cid, orders BY cid;

-- Left outer join: all customers, even those without orders
all_cust = JOIN customers BY cid LEFT OUTER, orders BY cid;

-- Result fields: customers::cid, customers::name, orders::oid, orders::amount
rich_data = FOREACH cust_orders GENERATE
 customers::name AS customer_name,
 orders::amount AS order_amount;

-- Reference data (small ~50MB lookup table)
countries = LOAD '/reference/countries.csv' AS (code:chararray, name:chararray);

-- Big sales data (100GB)
sales = LOAD '/data/sales/' AS (id:int, country_code:chararray, amount:float);

-- Replicated join: countries file goes into memory on every mapper
-- Much faster than shuffle join!
enriched = JOIN sales BY country_code, countries BY code USING 'replicated';

-- Sort ascending (default)
sorted_sales = ORDER sales BY amount;

-- Sort descending
top_sales = ORDER sales BY amount DESC;

-- Multi-column sort
sorted_multi = ORDER sales BY country ASC, amount DESC;

-- Usually combined with LIMIT for top-N analysis
top_10 = LIMIT (ORDER sales BY amount DESC) 10;

-- One relation → multiple relations based on conditions
SPLIT sales INTO
 premium IF amount >= 100000,
 standard IF amount >= 10000 AND amount < 100000,
 economy IF amount < 10000;

-- Now process each segment separately
STORE premium INTO '/output/premium' USING PigStorage(',');
STORE standard INTO '/output/standard' USING PigStorage(',');
STORE economy INTO '/output/economy' USING PigStorage(',');

-- Compare orders vs returns for each customer
orders = LOAD '/data/orders.csv' AS (cid:int, amount:float);
returns = LOAD '/data/returns.csv' AS (cid:int, reason:chararray);

-- Each output tuple: (cid, bag_of_orders, bag_of_returns)
comparison = COGROUP orders BY cid, returns BY cid;

-- Customers with more returns than orders
suspicious = FOREACH comparison
 GENERATE group AS cid,
 COUNT(orders) AS num_orders,
 COUNT(returns) AS num_returns
 WHERE COUNT(returns) > COUNT(orders);

-- Remove duplicates
unique_users = DISTINCT (FOREACH events GENERATE user);

-- Combine two relations (NOT deduplicating — use DISTINCT for that)
all_logs = UNION access_logs, error_logs;

-- Combine and deduplicate
all_unique = DISTINCT (UNION access_logs, error_logs);

-- STORE: permanent storage to HDFS (production use)
STORE results INTO '/output/analysis' USING PigStorage(',');

-- Store as Parquet (compressed columnar format)
STORE results INTO '/output/parquet_data'
 USING org.apache.pig.piggybank.storage.parquet.ParquetStorer();

-- DUMP: temporary output to console (debugging only)
DUMP results; -- Shows first few rows in terminal
-- Never use DUMP in production! It fetches ALL data to the client!

-- Pig doesn't execute ANYTHING until it absolutely must!
A = LOAD '/data/huge_file.csv' AS (...); -- Nothing happens yet
B = FILTER A BY amount > 1000; -- Still nothing
C = GROUP B BY country; -- Still nothing
D = FOREACH C GENERATE ...; -- Still nothing
STORE D INTO '/output/'; -- NOW it executes all steps!
 -- Pig optimizes the entire plan first!

-- See exactly what MapReduce jobs Pig will generate
EXPLAIN results;

-- Output shows:
-- Logical Plan: your operators as a tree
-- Physical Plan: how operators map to MapReduce primitives
-- MapReduce Plan: how many MR jobs, what runs in each

Database: analytics_db
 Table: sales
 Columns: id BIGINT, date STRING, amount DOUBLE, country STRING
 Partitions: country=India/dt=2026-02-26/ → hdfs:///data/sales/India/2026-02-26/
 Storage format: ORC
 Row count: 150,000,000
 Total size: 45 GB

 Table: customers
 Columns: cid BIGINT, name STRING, email STRING
 ...

Connect from Python:
from pyhive import hive
conn = hive.Connection(host='hiveserver2.company.com', port=10000, username='nitish')
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM sales')
print(cursor.fetchone()) # (150000000,)

Driver Step 1: Parse
 "Is this valid HiveQL syntax?" → Yes 
 AST (Abstract Syntax Tree) generated

Driver Step 2: Semantic Analysis 
 Consult Metastore: "Does table 'sales' exist?" → Yes 
 "Does column 'country' exist?" → Yes 
 "Does column 'amount' exist?" → Yes 

Driver Step 3: Query Optimization
 CBO (Cost-Based Optimizer): "How big is the table?" (from stats)
 "Should we use MapReduce, Tez, or Spark?" → Tez (faster)
 "Can we push the GROUP BY down?" → Yes, apply partition pruning

Driver Step 4: Generate Execution Plan
 MapReduce Stage 1:
 Map: Read sales, emit (country, amount)
 Reduce: Sum amounts per country
 Output to HDFS temp directory

Driver Step 5: Submit to YARN via Execution Engine

EXPLAIN SELECT country, SUM(amount) FROM sales GROUP BY country;

-- Configure the execution engine per session
SET hive.execution.engine = tez; -- Recommended for most queries
SET hive.execution.engine = spark; -- Best for complex analytics
SET hive.execution.engine = mr; -- Legacy, slowest

MAPREDUCE (3 chained jobs):
 HDFS Read → Map1 → Write to HDFS → Read → Map2 → Write HDFS → Read → Map3 → HDFS Write
 (Lots of slow disk I/O between stages!)

TEZ (same query as one DAG):
 HDFS Read → [Op1 → Op2 → Op3] → HDFS Write
 (Intermediate results stay in memory!)

# Beeline (modern, recommended)
beeline -u jdbc:hive2://hiveserver2:10000/default -n nitish -p password
beeline> SHOW DATABASES;
beeline> USE analytics;
beeline> SHOW TABLES;
beeline> DESCRIBE FORMATTED sales;

# Useful built-in commands
SET hive.exec.dynamic.partition=true; -- Allow dynamic partitioning
EXPLAIN SELECT ...; -- Show execution plan
ANALYZE TABLE sales COMPUTE STATISTICS; -- Update optimizer stats

-- MANAGED TABLE (Hive owns the data lifecycle)
CREATE TABLE transactions (
 id BIGINT,
 user_id BIGINT,
 amount DOUBLE,
 ts TIMESTAMP
)
STORED AS ORC;
-- DROP TABLE transactions → deletes BOTH metadata AND HDFS data! 

-- EXTERNAL TABLE (Hive is just a view over existing HDFS data)
CREATE EXTERNAL TABLE logs_external (
 ip STRING,
 url STRING,
 status INT,
 ts STRING
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
LOCATION '/data/weblogs/';
-- DROP TABLE logs_external → deletes ONLY metadata, HDFS data stays safe! 

-- Create a partitioned table (partitioned by date and country)
CREATE TABLE sales (
 id BIGINT,
 amount DOUBLE,
 product STRING
)
PARTITIONED BY (dt STRING, country STRING)
STORED AS ORC;

-- Load data into a specific partition
INSERT INTO sales PARTITION(dt='2026-02-26', country='India')
SELECT id, amount, product FROM staging_sales
WHERE dt='2026-02-26' AND country='India';

HDFS structure after partitioning:
 /warehouse/sales/
 dt=2026-01-01/
 country=India/ ← 500 MB
 country=US/ ← 1.2 GB
 country=UK/ ← 300 MB
 dt=2026-01-02/
 country=India/
 ...

Query: SELECT SUM(amount) FROM sales WHERE dt='2026-02-26' AND country='India'

WITHOUT partitioning: Scans ALL 500 GB of data 
WITH partitioning: Scans ONLY 500 MB of data (1000x less work!)

CREATE TABLE customers (
 cid BIGINT,
 name STRING,
 email STRING
)
CLUSTERED BY (cid) INTO 32 BUCKETS
STORED AS ORC;

CREATE TABLE orders (
 oid BIGINT,
 cid BIGINT,
 amount DOUBLE
)
CLUSTERED BY (cid) INTO 32 BUCKETS
STORED AS ORC;

-- Bucket map join: FAST because matching cid values are in the same bucket!
SELECT c.name, SUM(o.amount) as total_spend
FROM customers c
JOIN orders o ON c.cid = o.cid
GROUP BY c.name;

Without bucketing: With bucketing (32 buckets, JOIN on cid):
 Reducer must see ALL orders Bucket 1 of customers only needs
 and ALL customers to find bucket 1 of orders!
 matching cids. (1/32 of the data shuffled)

 Network: 100% of both tables Network: 3-10% of both tables 

-- Text (plain CSV/TSV) — worst performance but human readable
STORED AS TEXTFILE

-- ORC (Optimized Row Columnar) — best for Hive, fastest queries
STORED AS ORC

-- Parquet — best cross-platform (Spark, Presto, BigQuery all love it)
STORED AS PARQUET

-- Avro — best for schema evolution (adding/removing columns over time)
STORED AS AVRO

-- Sequence — Hadoop's binary key-value format
STORED AS SEQUENCEFILE

-- Running total of sales per country (ordered by date)
SELECT
 country,
 dt,
 amount,
 SUM(amount) OVER (
 PARTITION BY country
 ORDER BY dt
 ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
 ) AS cumulative_sales
FROM sales;

-- 4-day rolling average
SELECT
 country, dt, amount,
 AVG(amount) OVER (
 PARTITION BY country
 ORDER BY dt
 ROWS BETWEEN 3 PRECEDING AND CURRENT ROW
 ) AS rolling_4day_avg
FROM sales;

-- Rank customers by total spend
SELECT
 user_id,
 total_spend,
 RANK() OVER (ORDER BY total_spend DESC) AS spend_rank,
 DENSE_RANK() OVER (ORDER BY total_spend DESC) AS dense_rank,
 ROW_NUMBER() OVER (ORDER BY total_spend DESC) AS row_num,
 NTILE(4) OVER (ORDER BY total_spend DESC) AS quartile
FROM customer_totals;

-- Year-over-year comparison using LAG and LEAD
SELECT
 dt,
 revenue,
 LAG(revenue, 1) OVER (ORDER BY dt) AS prev_day_revenue,
 LEAD(revenue, 1) OVER (ORDER BY dt) AS next_day_revenue,
 revenue - LAG(revenue, 1) OVER (ORDER BY dt) AS day_over_day_change
FROM daily_revenue;

-- Enable (set in hive-site.xml or per session)
SET hive.support.concurrency=true;
SET hive.enforce.bucketing=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;

-- Create ACID-enabled table
CREATE TABLE inventory (
 product_id BIGINT,
 quantity INT,
 price DOUBLE
)
CLUSTERED BY (product_id) INTO 16 BUCKETS
STORED AS ORC
TBLPROPERTIES ('transactional'='true');

-- Now you can UPDATE and DELETE!
UPDATE inventory SET quantity = quantity - 10 WHERE product_id = 42;
DELETE FROM inventory WHERE quantity = 0;

-- MERGE for upsert operations (Slowly Changing Dimensions!)
MERGE INTO inventory AS target
USING new_stock AS source
ON target.product_id = source.product_id
WHEN MATCHED THEN
 UPDATE SET quantity = target.quantity + source.incoming_qty
WHEN NOT MATCHED THEN
 INSERT VALUES (source.product_id, source.incoming_qty, source.price);

-- 1. Always check your execution plan first!
EXPLAIN SELECT country, SUM(amount) FROM sales GROUP BY country;

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

-- 3. Keep table statistics up to date
ANALYZE TABLE sales COMPUTE STATISTICS;
ANALYZE TABLE sales COMPUTE STATISTICS FOR COLUMNS;

-- 4. Use partition pruning (always filter on partition columns!)
SELECT * FROM sales WHERE dt = '2026-02-26'; -- Fast (partition pruned!)
SELECT * FROM sales WHERE YEAR(dt) = 2026; -- Slow (full scan, can't prune)

-- 5. Enable vectorized execution
SET hive.vectorized.execution.enabled=true;

-- 6. Use Tez or Spark instead of MapReduce
SET hive.execution.engine=tez;

Row keys: 
 2026-02-26-10:00:00 → sensor reading
 2026-02-26-10:00:01 → sensor reading
 2026-02-26-10:00:02 → sensor reading

Problem: All writes go to the SAME region (the latest one)!
This is called HOTSPOTTING — one server gets all the traffic!

Solution: Add a hash prefix to distribute load!

Original key: 2026-02-26:device_123
Salted key: a7:2026-02-26:device_123 ← first 2 chars = hash of key % 16

Now writes distribute across 16 regions instead of 1! 

Client writes: PUT row_key="user:001", col="CF:name", value="Riya"

Step 1: Write to WAL (Write-Ahead Log) on disk first!
 (If server crashes before completing write,
 WAL lets you recover!)

Step 2: Write to MemStore (in-memory write buffer)
 Acknowledgment sent to client: "Write successful!" 

Step 3: When MemStore is full (~64MB default):
 FLUSH → Write MemStore to new HFile on HDFS
 MemStore cleared → ready for more writes!

Step 4: Over time, many small HFiles accumulate.
 COMPACTION: Merge small HFiles → fewer large HFiles
 (Minor compaction: merge few small files)
 (Major compaction: merge ALL files, remove deleted cells)

Client reads: GET row_key="user:001", col="CF:name"

Step 1: Check MemStore (most recent writes, in memory)
 Found? Return value immediately! 

Step 2: Check BlockCache (recently read HFile blocks in memory)
 Found? Return value! 

Step 3: Read from HFiles on HDFS (disk access - slower)
 Read all relevant HFile blocks for this row key
 Return most recent version!

Optimization: BloomFilters on each HFile tell whether a row
 key might exist in it (avoid unnecessary disk reads!)

/ (root)
├── /hbase
│ ├── /hbase/master → "hm1.company.com" (who is active HMaster?)
│ ├── /hbase/rs → List of alive RegionServers
│ │ ├── /hbase/rs/rs1.host → (ephemeral → disappears when RS dies!)
│ │ ├── /hbase/rs/rs2.host
│ │ └── /hbase/rs/rs3.host
│ └── /hbase/regions → Which RS owns which region?
├── /kafka
│ └── /kafka/controller → "broker_3" (who is Kafka controller?)
└── /yarn
 └── /yarn/leader → "rm1.company.com" (active ResourceManager)

When RegionServer rs3 starts:
 Creates /hbase/rs/rs3.host (ephemeral znode!)

When rs3 CRASHES (without cleanup):
 ZooKeeper detects session timeout
 AUTOMATICALLY deletes /hbase/rs/rs3.host

All watchers (HMaster, clients) get notified: "rs3 is gone!"
HMaster: "Ok, reassign rs3's regions to other RegionServers!" 

All HMaster candidates race to create /hbase/master (ephemeral node)

Candidate 1: "I'll create /hbase/master!" → SUCCESS → Becomes ACTIVE! 
Candidate 2: "I'll create /hbase/master!" → FAILS (already exists) → Becomes STANDBY
Candidate 3: → FAILS → Becomes STANDBY

All standbys WATCH /hbase/master
If active HMaster dies → /hbase/master deleted → race starts again!

Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "zk1,zk2,zk3");

try (Connection connection = ConnectionFactory.createConnection(conf);
 Table table = connection.getTable(TableName.valueOf("sensors"))) {

 // ─── PUT (Write) ─────────────────────────────────────────────
 Put put = new Put(Bytes.toBytes("device:001:20260226100000")); // row key
 put.addColumn(
 Bytes.toBytes("readings"), // column family
 Bytes.toBytes("temperature"),// qualifier
 Bytes.toBytes("28.5") // value
 );
 put.addColumn(
 Bytes.toBytes("readings"),
 Bytes.toBytes("humidity"),
 Bytes.toBytes("65.2")
 );
 table.put(put);

 // ─── GET (Read single row) ────────────────────────────────────
 Get get = new Get(Bytes.toBytes("device:001:20260226100000"));
 Result result = table.get(get);
 byte[] temp = result.getValue(
 Bytes.toBytes("readings"),
 Bytes.toBytes("temperature")
 );
 System.out.println("Temperature: " + Bytes.toString(temp)); // 28.5

 // ─── SCAN (Range query) ───────────────────────────────────────
 Scan scan = new Scan();
 scan.setStartRow(Bytes.toBytes("device:001:20260226090000")); // 9 AM
 scan.setStopRow(Bytes.toBytes("device:001:20260226110000")); // 11 AM
 scan.addFamily(Bytes.toBytes("readings"));

 ResultScanner scanner = table.getScanner(scan);
 for (Result r : scanner) {
 System.out.println("Row: " + Bytes.toString(r.getRow()));
 }
}

Traditional workflow (technical):
 Write HiveQL → Submit to cluster → Wait 10 min → Get CSV → Open Excel → Make chart

BigSheets workflow (non-technical):
 Open BigSheets UI → Browse HDFS files (like Google Drive) → 
 Point and click column selection → Click "Chart" button → Done! 

# BigInsights Text Analytics API example
# Extract sentiment from customer feedback stored in HDFS
bigsql> SELECT review_text,
 SENTIMENT(review_text) as sentiment,
 EXTRACT_KEYWORDS(review_text) as keywords
 FROM customer_reviews
 WHERE product_id = 42;

// A simple stock market alert system in SPL

// Step 1: Read from market data socket
stream<rstring symbol, float64 price, uint64 volume>
 MarketFeed = ReadFromSocket() {
 param
 host: "feeds.nasdaq.com";
 port: 9090;
 format: csv;
 }

// Step 2: Filter only valid prices
stream<rstring symbol, float64 price, uint64 volume>
 ValidData = Filter(MarketFeed) {
 param
 filter: price > 0.0 && volume > 0ul;
 }

// Step 3: Compute 1-second average price (tumbling window)
stream<rstring symbol, float64 avgPrice>
 Aggregated = Aggregate(ValidData) {
 window
 ValidData: tumbling, time(1.0); // 1-second window
 param
 groupBy: symbol;
 output
 Aggregated: avgPrice = Average(price);
 }

// Step 4: Detect threshold crossings
stream<rstring symbol, float64 avgPrice, rstring alert>
 Alerts = Functor(Aggregated) {
 param
 filter: avgPrice > 5000.0; // Alert if price > ₹5000
 output
 Alerts: alert = "PRICE_SPIKE: " + symbol;
 }

// Step 5: Send alerts to Kafka
() as AlertSink = PublishKafka(Alerts) {
 param
 topic: "price-alerts";
 brokers: "kafka1:9092,kafka2:9092";
}

BigInsights:
 Store 5 years of historical tick data on HDFS (petabytes)
 Run overnight batch jobs to backtest trading strategies

IBM Streams:
 Process live market feeds in < 5ms
 Detect arbitrage opportunities across exchanges
 Trigger automated trades when conditions met

Together: Historical analysis informs the model, real-time execution makes the trades!

BigInsights:
 Store all CDRs (Call Detail Records) — billions per day
 Run monthly fraud pattern analysis on historic data

IBM Streams:
 Analyze live CDR stream as calls happen
 Detect if same number calling 50 different victims within 1 minute
 Block the number in < 10ms (before the 51st call!)

Together: Batch finds patterns, Streams acts on them instantly!

small_table = LOAD '/reference/lookup.csv' AS (...); -- 5 MB
big_log = LOAD '/data/transactions/' AS (...); -- 200 GB

result = JOIN big_log BY key, small_table BY key USING 'replicated';
-- small_table is broadcast to every mapper!

Original: 20260226-100000-device123
Salted: a7:20260226-100000-device123 (hash=a7, distributes to different regions)