Pig, Hive, and Data Transformation Frameworks

# Pig, Hive, and Data Transformation Frameworks

6.1 Lecture Overview, Progress Tracking, and Hadoop Ecosystem Architecture

6.1.1 Hadoop Curriculum Positioning and Core Infrastructure Stack

Enterprise big data architectures rely on a layered stack of storage engines, distributed execution frameworks, multi-tenant schedulers, and data transformation interfaces. In the overall big data ecosystem maturity model, foundational infrastructure comprises two core capabilities:

• Infinite Distributed Storage: Provided by the Hadoop Distributed File System (HDFS), which splits large datasets into fixed-size blocks (default 128 MB each) and distributes replica copies across a cluster of commodity machines. A master NameNode tracks which blocks live on which DataNodes, while the DataNodes store the actual data and report heartbeats to the master.

• Infinite Parallel Processing: Provided by the MapReduce (MR) programming paradigm, which executes data-localized map, shuffle, sort, and reduce operations across multi-node clusters. The key insight is data locality — moving computation to where the data resides rather than shipping data across the network to a central processor.

Having established how HDFS manages block storage and how MapReduce parallelizes key-value computations, the current curriculum advances to data flow scripting, data warehousing, and cluster coordination. Specifically, this lecture covers Apache Pig (a high-level data flow scripting language created by Yahoo) and introduces Apache Hive (a data warehousing infrastructure created by Facebook). These components eliminate the need to write raw Java MapReduce programs. Subsequent course modules progress to cluster coordination using Apache Zookeeper, NoSQL database architectures (including Apache HBase and Apache Cassandra), followed by Apache Spark and distributed machine learning after the mid-semester evaluation.

6.1.2 Recap of Lecture 5: YARN Scheduling, Sqoop Ingestion, and Flume Log Streaming

To contextualize high-level data processing frameworks, consider the core infrastructure components established in previous lectures:

• YARN (Yet Another Resource Negotiator): Operates as a master-slave architecture decoupling cluster resource management from application execution. The master Node hosts the Resource Manager, which contains two sub-components:

1. Application Manager: Accepts incoming job submissions, negotiates initial resource containers, and spins off an Application Master instance on a slave node. Think of the Application Manager as a receptionist — it takes your request and assigns you a coordinator.
2. Scheduler: Allocates cluster memory and CPU containers across competing applications according to configured scheduling policies:

• FIFO Scheduler — first-in, first-out; simplest but can starve small jobs behind large ones
• Capacity Scheduler — guarantees a minimum capacity to each queue (used by Yahoo)
• Fair Scheduler — dynamically balances resources so all applications get roughly equal share (used by Facebook)

• Slave Nodes: Run NodeManager daemons responsible for monitoring container resource utilization and launching actual map and reduce tasks. Each NodeManager reports available resources to the Resource Manager via periodic heartbeats.

• Sqoop (SQL-to-Hadoop): A relational data ingestion tool designed to import and export structured datasets bi-directionally between Relational Database Management Systems (RDBMS) such as MySQL, Oracle, or PostgreSQL and HDFS or Hive tables. Sqoop generates MapReduce jobs under the hood to perform the data transfer in parallel, reading from and writing to multiple database partitions simultaneously.

• Flume: A distributed, reliable event collection service used to stream unstructured log data (such as web server logs or streaming network packets) into HDFS logger sinks. A Flume agent has three components:
• Source — receives data (e.g., Netcat, Avro, exec, Spooling Directory)
• Channel — buffers data (memory channel for speed, file channel for durability)
• Sink — writes data to the destination (HDFS, HBase, Logger)

6.1.3 The MapReduce Usability Gap and Motivation for High-Level Frameworks

Although MapReduce provides massive computational scalability, writing native MapReduce applications in Java presents a steep development barrier. Developers must manually implement low-level Java code comprising three distinct components:

1. Job Configuration: Writing verbose Java boilerplate to configure input/output formats, key-value serialization classes (Writable), mapper classes, reducer classes, and cluster job parameters.
2. Mapper Class: Deconstructing business logic into parallel map steps that process raw input key-value pairs and emit intermediate key-value pairs. Each mapper must implement the Mapper<K1, V1, K2, V2> interface and override the map() method.
3. Reducer Class: Defining shuffle-and-sort aggregation logic to process groups of intermediate values associated with each key. Each reducer must implement the Reducer<K2, V2, K3, V3> and override the reduce() method.

lines = LOAD 'input/text.txt' AS (line:chararray);
words = FOREACH lines GENERATE FLATTEN(TOKENIZE(line)) AS word;
grouped = GROUP words BY word;
word_count = FOREACH grouped GENERATE group AS word, COUNT(words) AS count;
DUMP word_count;

Back in 2004, Java and MapReduce were new technologies in the enterprise ecosystem. Engineering teams at major tech companies like Yahoo and Facebook possessed vast datasets and legacy workflows built heavily around SQL and relational databases. Expecting entire organizations of data analysts and DBAs to transition from SQL to writing complex Java MapReduce code created severe operational friction. To bridge this usability gap, Yahoo and Facebook independently developed high-level abstractions on top of MapReduce:

• Facebook created Apache Hive to provide a familiar SQL-like query interface (HiveQL) over HDFS data — aimed at analysts and BI tools that already spoke SQL.
• Yahoo created Apache Pig to provide a procedural, data-flow scripting language (Pig Latin) for complex data transformations — aimed at engineers building ETL pipelines.

records = LOAD 'weather.txt' AS (year:chararray, temp:int, quality:int);
filtered = FILTER records BY temp != 9999 AND quality IN (0,1,4,5,9);
grouped = GROUP filtered BY year;
max_temp = FOREACH grouped GENERATE group, MAX(filtered.temp);
DUMP max_temp;

SELECT year, MAX(temperature) FROM records
WHERE temperature != 9999 AND quality IN (0,1,4,5,9)
GROUP BY year;

Real-world: Modern data platforms leverage high-level abstractions like Hive and Pig (and subsequently Spark SQL) to allow data engineers and business analysts to query petabyte-scale data lakes without writing raw Java MapReduce code.

6.1.4 Student Questions and Answers

6.2 Apache Pig Fundamentals, Evolution, and Data Flow Scripting

6.2.1 Origins at Yahoo and Twitter Industry Adoption

Apache Pig was originally developed at Yahoo around 2006 to provide an efficient, high-level data flow environment for processing large-scale datasets on Hadoop clusters. The project was led by Yahoo Research and aimed squarely at a practical problem: the vast majority of Yahoo's data engineers and researchers knew SQL and scripting languages, not Java. Writing ETL pipelines in raw MapReduce was blocking productivity.

At Yahoo, engineering teams used Pig to write ETL (Extract, Transform, Load) pipelines, web crawl analysis scripts, and search indexing jobs. Pig became so central to Yahoo's data operations that the company open-sourced it in 2007 under the Apache Software Foundation.

At its peak between 2012 and 2017, Apache Pig achieved massive adoption across the tech industry:

• Yahoo reported that over 80% of all MapReduce jobs executed across its multi-thousand node Hadoop clusters were generated by Pig Latin scripts.
• Twitter heavily adopted Apache Pig for log processing, user engagement analytics, and graph mining algorithms.
• Other major adopters included LinkedIn, Netflix, and Nokia.

6.2.2 Pig Latin Language Design and Productivity Gains over Java MapReduce

Pig Latin is a procedural data-flow language designed specifically for data manipulation. Unlike declarative SQL (which describes what data to retrieve without specifying execution order), Pig Latin allows developers to write an explicit sequence of data transformation steps (how to transform data step by step).

Pig Latin scripts offer massive productivity advantages over raw Java MapReduce:

Pedagogical Analogy: Writing Java MapReduce compared to Pig Latin is like writing low-level C code (where developers manually allocate memory and build data structures from scratch) versus writing modern Python scripts (where high-level built-in data types and operations reduce complex algorithms to concise single-line expressions).

sales = LOAD 'sales.csv' USING PigStorage(',') AS (region:chararray, revenue:double);
filtered = FILTER sales BY revenue > 50000.0;
grouped = GROUP filtered BY region;
avg_rev = FOREACH grouped GENERATE group AS region, AVG(filtered.revenue) AS avg_revenue;
DUMP avg_rev;

6.2.3 Ecosystem Transition: Pig Decline and Spark Paradigm Shift

The big data ecosystem evolves rapidly as processing demands change. Between 2012 and 2015, Apache Pig represented the industry standard for Hadoop data processing. However, over time, the popularity of Pig declined due to the emergence of Apache Spark.

Apache Spark introduced in-memory cluster computing and resilient distributed datasets (RDDs), performing data transformations orders of magnitude faster than disk-based MapReduce engines. The key performance difference:

• Pig (MapReduce-based): Every intermediate result is written to HDFS disk between MapReduce stages. For a multi-step Pig script, this means repeated disk I/O.
• Spark: Intermediate results can be kept in memory (RAM) across stages, eliminating disk writes for iterative or multi-step computations.

While Apache Pig's usage shrank as organizations migrated to Spark, Apache Hive maintained longer longevity due to its direct SQL compatibility with legacy enterprise reporting tools.

6.2.4 Student Questions and Answers

6.3 Pig Execution Modes, Architecture, and Grunt Shell CLI

6.3.1 Local Mode versus MapReduce Cluster Mode Execution

When launching Apache Pig, developers can choose between two operational execution modes depending on whether they are developing scripts locally or executing production workloads on Hadoop:

6.3.2 Structural Limitations: Absence of Web Server, Storage Engine, and OLTP Capabilities

To evaluate where Apache Pig fits into enterprise architectures, it is essential to understand what Pig cannot do:

1. No Native Storage Engine: Unlike relational databases (such as MySQL or PostgreSQL) which manage dedicated physical storage files on disk, Pig possesses zero native storage capabilities. Pig is purely a client-side execution framework that reads data stored externally in HDFS or local storage. Pig does not "own" any data — it only transforms it.

1. No Background Daemon or Web Server: Pig does not run a persistent background server daemon or SQL server process. External applications written in C++, Java, or Python cannot connect remotely to Pig via JDBC/ODBC network daemons. Access occurs exclusively through the command-line client.

1. Not Suitable for Real-Time Processing or OLTP: Pig scripts compile into batch MapReduce jobs. Because Hadoop MapReduce involves YARN container allocation, map-reduce task initialization, and disk I/O serialization, job latency ranges from tens of seconds to hours. Pig cannot be used for Online Transaction Processing (OLTP) or sub-second real-time analytics.

1. Schema-on-Read / Schema-Less Flexibility: Pig does not enforce rigid schema constraints upon loading. A CSV file can be loaded with or without explicit column names and data types. This is schema-on-read — the schema is applied when the data is read, not when it is written.

1. Ineffective for Unstructured Data: Pig handles structured and semi-structured datasets (such as CSV, TSV, or JSON) effectively, but is poorly suited for unstructured audio, video, or arbitrary binary blobs.

6.3.3 Grunt Shell CLI and Interactive Script Execution

Apache Pig provides an interactive Command Line Interface (CLI) known as the Grunt Shell. Upon executing the pig command in the terminal, the environment initializes the Grunt prompt:

grunt>

From the Grunt shell, developers can interactively type Pig Latin statements line by line, inspect intermediate relations using DESCRIBE or DUMP, debug transformation logic on sample records, and execute full batch scripts (.pig files).

6.3.4 Etymology and Pedagogical Analogy of Apache Pig

The naming of Apache Pig follows a lighthearted animal naming convention common across the Hadoop ecosystem (such as Hadoop's yellow elephant mascot or Mahout's elephant keeper). According to official Apache Pig documentation, the name "Pig" carries three thematic justifications:

• "Pigs eat anything": Apache Pig can ingest any structured or semi-structured data format regardless of schema (CSV, TSV, JSON, custom formats).
• "Pigs live anywhere": Pig Latin scripts run anywhere across the Hadoop cluster ecosystem (local mode, pseudo-distributed, fully distributed).
• "Pigs are domestic and adaptable": Pig provides a flexible, developer-friendly environment for arbitrary data transformation tasks, including User-Defined Functions (UDFs) for custom logic.

Real-world: Pig is classified strictly as an ETL (Extract, Transform, Load) data transformation tool. It is used to ingest raw incoming data, clean corrupted records, join datasets, calculate computed fields, and output refined files ready for data warehouse consumption.

6.3.5 Student Questions and Answers

6.4 Pig Data Model and Data Types

6.4.1 Primitive Data Types and Type System

Apache Pig supports standard primitive data types common across programming languages:

6.4.2 Complex Data Structures: Tuples, Bags, and Maps

To handle complex, nested data representations, Pig Latin defines three constructible complex data structures:

1. Tuple

A Tuple is an ordered sequence of fields. Fields inside a tuple can be of any data type (primitive or complex). Tuples are enclosed in standard round parentheses (...) with comma-separated values:

\[ \text{Tuple Representation:} \quad (v_1, v_2, v_3, \dots, v_k) \]

Verbal description: A tuple represents an ordered list of fields enclosed in round parentheses, analogous to a single database row. Each field has a position (zero-indexed) and can be accessed by position (\$0, \$1, \$2) or by name if a schema is defined.

Example:

(1, 'Europe', 'Snacks', 4876, 97.44, 152.58)

2. Bag

A Bag is an unordered collection of tuples. Duplicate tuples are permitted within a bag. Bags are enclosed in curly braces {...} containing comma-separated tuples:

\[ \text{Bag Representation:} \quad \{ t_1, t_2, t_3, \dots, t_n \} = \{ (v_{11}, v_{12}), (v_{21}, v_{22}), \dots \} \]

Verbal description: A bag is an unordered collection of tuples enclosed in curly braces, analogous to an unindexed database table or relation partition. The GROUP operator produces a bag for each group key.

Example:

{ (1, 'Europe', 'Snacks'), (2, 'Asia', 'Milk'), (3, 'Asia', 'Groceries') }

3. Map

A Map is a collection of key-value pairs, where each key must be a unique string (chararray) and the value can be of any data type. Maps are enclosed in square brackets [...], with keys and values separated by a hash # symbol:

\[ \text{Map Representation:} \quad [ k_1 \# v_1, k_2 \# v_2, \dots, k_m \# v_m ] \]

Verbal description: A map is a set of key-value pairs enclosed in square brackets, where key and value are separated by a hash character. Keys are always chararray (strings); values can be any type.

Example:

[ 'id'#1, 'name'#'alpha', 'salary'#10000 ]

6.4.3 Mathematical and Structural Representations of Complex Types

In formal database terms, Pig's complex data types correspond directly to relational algebra components:

• A relation \(R\) in Pig Latin is represented as a Bag of Tuples:

\[ R = \{ t \mid t = (x_1, x_2, \dots, x_d) \} \] where each element \(x_i\) belongs to domain \(D_i\) (the data type of field \(i\)).

• When a dataset is grouped by a key attribute \(k\), Pig constructs a grouped relation containing nested bags:

\[ R_{\text{grouped}} = \{ (k_A, \{ t \in R \mid t.\text{key} = k_A \}), (k_B, \{ t \in R \mid t.\text{key} = k_B \}) \} \] where each unique key \(k_A\) is paired with a nested bag containing all matching tuples.

R = { (1, 'Europe', 743980.08), (2, 'Asia', 440000.00), (3, 'Europe', 5286522.78) }

R_grouped = {
  ('Europe', { (1, 'Europe', 743980.08), (3, 'Europe', 5286522.78) }),
  ('Asia',   { (2, 'Asia', 440000.00) })
}

6.4.4 Student Questions and Answers

6.5 Pig Data Transformation Language: Loading, Inspected Data, and Output

6.5.1 The LOAD Statement and Positional versus Named Schemas

To bring external file data into Apache Pig for processing, developers use the LOAD statement. Executing a LOAD statement defines a Relation (the Pig term for a tabular dataset handle — a name you use to refer to the data in subsequent operations).

Positional Loading (No Schema)

If data is loaded without column definitions:

sales = LOAD 'sales.csv';

Pig assigns default positional variables to fields: \$0 for the first column, \$1 for the second column, \$2 for the third column, and so on. All fields default to bytearray type.

Named Loading with Explicit Schema

Developers can specify column names and primitive data types using the AS clause:

sales = LOAD 'sales.csv' AS (
    index: int,
    region: chararray,
    item_type: chararray,
    units_sold: int,
    unit_cost: float,
    unit_price: float
);

Once loaded with a schema, fields can be referenced directly by name (e.g., units_sold) or by positional offset (\$3). Named references are clearer and less error-prone.

6.5.2 Delimiter Handling using PigStorage

By default, Pig expects input files to be tab-delimited (\t). When loading files delimited by other characters (such as comma-separated CSV files), developers must specify the PigStorage deserializer using the USING clause:

sales = LOAD 'sales.csv' 
USING PigStorage(',') 
AS (
    index: int,
    region: chararray,
    item_type: chararray,
    units_sold: int,
    unit_cost: float,
    unit_price: float
);

6.5.3 Inspecting Relations with DESCRIBE and DUMP

Pig Latin provides diagnostic operators to inspect relation schemas and data contents during Grunt shell sessions:

1. DESCRIBE relation_name;: Displays the schema, field names, and data types of a relation without scanning file data (instant — no MapReduce job):

grunt> DESCRIBE sales;
-- Output: sales: {index: int, region: chararray, item_type: chararray, units_sold: int, unit_cost: float, unit_price: float}

1. DUMP relation_name;: Triggers MapReduce job execution, reads relation data, and prints the formatted bag of tuples directly to the terminal screen:

grunt> DUMP sales;
-- Output:
-- (1,Europe,Snacks,4876,97.44,152.58)
-- (2,Central America and the Caribbean,Baby Food,2921,159.42,255.28)
-- (3,Europe,Office Supplies,8118,524.96,651.21)

6.5.4 Storing Transformed Data into Folders using STORE

To persist a processed relation to storage, developers use the STORE statement:

STORE sales INTO '/output/sales_processed' USING PigStorage(',');

6.5.5 Worked Example: Complete Ingestion Pipeline on Sales Dataset

1,Australia and Oceania,Snacks,4876,97.44,152.58
2,Central America and the Caribbean,Baby Food,2921,159.42,255.28
3,Europe,Office Supplies,8118,524.96,651.21

-- Step 1: Load CSV data with explicit schema using PigStorage
sales = LOAD 'sales.csv' 
USING PigStorage(',') 
AS (
    index: int,
    region: chararray,
    item_type: chararray,
    units_sold: int,
    unit_cost: float,
    unit_price: float
);

-- Step 2: Print schema to confirm field mapping
DESCRIBE sales;
-- Output: sales: {index: int, region: chararray, item_type: chararray, units_sold: int, unit_cost: float, unit_price: float}

-- Step 3: Write processed relation to HDFS output directory
STORE sales INTO 'output/sales_ingested' USING PigStorage(',');

6.5.6 Student Questions and Answers

6.6 Relational & Flow Operators in Pig Latin

6.6.1 Projection and Derived Column Creation with FOR EACH ... GENERATE

The FOR EACH ... GENERATE statement iterates row-by-row over a relation, projecting specific columns or creating new calculated attributes. Think of it as saying: "For each row in this dataset, generate a new row with these fields."

Column Projection

To extract a subset of columns (such as index, region, and item_type):

trimmed_sales = FOREACH sales GENERATE index, region, item_type;

Or using positional offsets:

trimmed_sales = FOREACH sales GENERATE \$0, \$1, \$2;

Positional Range Shortcuts

Pig Latin supports positional range shortcuts to simplify column selection:

• \$0..\$5: Selects all columns from index \$0 through index \$5 inclusive.
• ..\$5: Selects all columns from the first field up to index \$5.
• \$3..: Selects all columns from index \$3 through the final column of the relation.

6.6.2 Mathematical Calculations: Revenue and Profit Formulas

Developers can compute new mathematical fields within FOR EACH ... GENERATE projections.

1. Revenue Calculation

Revenue for a sales row is calculated as units sold multiplied by selling price per unit:

\[ \text{Revenue} = \text{Units Sold} \times \text{Selling Price} \]

Verbal description audit trail: Units sold multiplied by selling cost gives the total revenue for that row.

Mathematical formulation: \[ R_i = U_i \times P_i \] where \(R_i \in \mathbb{R}_{\ge 0}\) is the row revenue, \(U_i \in \mathbb{Z}_{\ge 0}\) is the number of units sold, and \(P_i \in \mathbb{R}_{\ge 0}\) is the selling price per unit.

Pig Latin implementation:

sales_with_revenue = FOREACH sales GENERATE 
    \$0..\$5, 
    (units_sold * unit_price) AS revenue: float;

2. Profit Calculation

Profit per row is calculated as units sold multiplied by the difference between selling price and unit cost:

\[ \text{Profit} = \text{Units Sold} \times (\text{Selling Price} - \text{Unit Cost}) \]

Verbal description audit trail: Selling price minus cost price gives unit profit, multiplied by units sold gives total profit.

Mathematical formulation: \[ \Pi_i = U_i \times (P_i - C_i) \] where \(\Pi_i \in \mathbb{R}\) is the total profit for row \(i\), and \(C_i \in \mathbb{R}_{\ge 0}\) is the unit manufacturing cost.

Pig Latin implementation:

sales_with_profit = FOREACH sales GENERATE 
    \$0..\$5, 
    (units_sold * (unit_price - unit_cost)) AS profit: float;

6.6.3 Row Filtering with FILTER (Numeric, Comparison, and Regex)

The FILTER operator selects tuples from a relation that satisfy a logical condition (equivalent to the SQL WHERE clause). Only tuples for which the condition evaluates to true are retained.

Numeric Filtering To select records where revenue exceeds \$100,000:

high_revenue_sales = FILTER sales_with_revenue BY revenue > 100000.0;

String and Regular Expression Filtering To select records where item_type contains the substring 'Baby':

baby_products = FILTER sales BY item_type MATCHES '.*Baby.*';

6.6.4 Grouping with GROUP BY and Nested Bag Formats

The GROUP operator collects all tuples in a relation that share identical values for a specified grouping key.

grouped_by_region = GROUP sales_with_revenue BY region;

Output Structure of GROUP

When a relation is grouped, Pig transforms the flat relation into a new relation containing two top-level fields:

1. group: The key value shared by the group (e.g., 'Europe').
2. A nested Bag named after the original relation (sales_with_revenue), containing all full tuples matching that key.

Schema structure produced by GROUP:

grouped_by_region: {
    group: chararray, 
    sales_with_revenue: {
        (index: int, region: chararray, item_type: chararray, units_sold: int, unit_cost: float, unit_price: float, revenue: float)
    }
}

DUMP visual output snippet:

('Europe', { (1, 'Europe', 'Snacks', 4876, 97.44, 152.58, 743980.08), (3, 'Europe', 'Office Supplies', 8118, 524.96, 651.21, 5286522.78) })
('Asia', { (2, 'Asia', 'Milk', 2000, 130.00, 220.00, 440000.00) })

6.6.5 Grouped Aggregations: Average, Sum, Min, Max

Once data is grouped, developers combine FOR EACH ... GENERATE with built-in aggregate functions (AVG, SUM, MIN, MAX, COUNT) to compute summary metrics per group.

To compute average revenue per region:

\[ \text{AVG}(\text{Revenue}_{\text{region}}) = \frac{1}{N_{\text{region}}} \sum_{j=1}^{N_{\text{region}}} R_{j,\text{region}} \]

where \(N_{\text{region}}\) is the number of tuples in the grouped bag for that region.

avg_revenue_per_region = FOREACH grouped_by_region GENERATE 
    group AS region, 
    AVG(sales_with_revenue.revenue) AS avg_revenue: double;

6.6.6 Relational Operations: JOIN, ORDER BY, LIMIT, DISTINCT, SPLIT, SAMPLE, UNION, and COGROUP

1. JOIN

Combines tuples from two or more relations based on matching key values:

joined_data = JOIN relation1 BY id, relation2 BY customer_id;

Supports LEFT OUTER, RIGHT OUTER, and FULL OUTER joins. By default, JOIN performs an inner join (only matching rows are kept).

2. ORDER BY

Sorts a relation based on one or more sort keys in ascending (ASC, the default) or descending (DESC) order:

sorted_sales = ORDER sales_with_revenue BY revenue DESC;

3. LIMIT

Restricts the output relation to a specified maximum number of tuples:

top_10_sales = LIMIT sorted_sales 10;

4. DISTINCT

Removes duplicate tuples from a relation (operates on the entire tuple, not individual fields):

unique_regions = DISTINCT (FOREACH sales GENERATE region);

5. SPLIT

Partitions a single relation into two or more relations based on conditional expressions. Conditions do not need to be mutually exclusive (a tuple can appear in multiple output relations):

SPLIT sales_with_revenue INTO 
    low_revenue IF revenue < 30000.0,
    medium_revenue IF (revenue >= 30000.0 AND revenue < 100000.0),
    high_revenue IF revenue >= 100000.0;

6. SAMPLE

Extracts a random sample fraction \(p \in [0.0, 1.0]\) of tuples from a relation:

sampled_sales = SAMPLE sales 0.1; -- Extracts approximately a 10% random sample

7. UNION

Combines the contents of two or more relations into a single relation (like SQL UNION ALL — duplicates are kept):

all_sales = UNION relation1, relation2;

8. COGROUP

Groups tuples from two or more relations sharing common keys, producing independent nested bags for each input relation:

cogrouped_data = COGROUP customers BY id, orders BY customer_id;

6.6.7 Pitfalls: Non-Deterministic LIMIT without ORDER BY

If a developer executes:

-- INCORRECT TOP 10 PIPELINE
top_10_raw = LIMIT sales_with_revenue 10;

Pig will return an arbitrary, non-deterministic set of 10 records based on whichever YARN mapper finished reading file splits first. To reliably extract top records, developers must sort the dataset with ORDER BY prior to calling LIMIT:

-- CORRECT TOP 10 PIPELINE
sorted_sales = ORDER sales_with_revenue BY revenue DESC;
top_10_revenue = LIMIT sorted_sales 10;

6.6.8 Worked Walkthrough: End-to-End Sales Data Processing Script

-- Step 1: Load raw sales data
sales = LOAD 'sales.csv' 
USING PigStorage(',') 
AS (
    index: int,
    region: chararray,
    item_type: chararray,
    units_sold: int,
    unit_cost: float,
    unit_price: float
);

-- Step 2: Compute calculated revenue field
sales_rev = FOREACH sales GENERATE 
    region, 
    item_type, 
    units_sold, 
    (units_sold * unit_price) AS revenue: float;

-- Step 3: Filter records with revenue over \$50,000
filtered_sales = FILTER sales_rev BY revenue > 50000.0;

-- Step 4: Group filtered sales by region
grouped_sales = GROUP filtered_sales BY region;

-- Step 5: Compute average revenue per region
regional_summary = FOREACH grouped_sales GENERATE 
    group AS region, 
    AVG(filtered_sales.revenue) AS avg_revenue: double;

-- Step 6: Order summary by average revenue descending
sorted_summary = ORDER regional_summary BY avg_revenue DESC;

-- Step 7: Limit output to top 5 regions
top_5_regions = LIMIT sorted_summary 5;

-- Step 8: Store output to HDFS
STORE top_5_regions INTO 'output/top_5_regional_revenue' USING PigStorage(',');

6.6.9 Student Questions and Answers

6.7 Built-In Functions and User-Defined Functions (UDFs) in Pig

6.7.1 Built-in Date, Time, and Statistical Utilities

Apache Pig provides extensive libraries of built-in functions across several categories:

• Date and Time Functions: AddDuration, CurrentTime, DaysBetween, GetYear, GetMonth, GetDay, ToDate. Useful for time-series analysis and log processing.
• Statistical/Math Functions: AVG, SUM, MIN, MAX, COUNT, COUNT_STAR, STDEV, VARIANCE, ABS, CEIL, FLOOR, ROUND, SQRT, LOG.
• String Functions: CONCAT, SUBSTRING, LOWER, UPPER, TRIM, INDEXOF, REPLACE, STRSPLIT, TOKENIZE, SIZE.
• Conversion Functions: TOTUPLE, TOBAG, TOMAP, TOP, FLATTEN.
• I/O Functions: PigStorage (delimited text), JsonLoader/JsonStorage (JSON), AvroStorage (Avro), ParquetLoader/ParquetStorer (Parquet), OrcStorage (ORC).

6.7.2 Extending Pig with Custom Java UDFs

When complex business logic (such as proprietary encryption algorithms, custom statistical modeling, or domain-specific differential calculations) cannot be expressed using built-in Pig functions, developers can extend Pig using User-Defined Functions (UDFs).

-- Step 1: Register the compiled custom Java UDF JAR
REGISTER /path/to/custom_udf.jar;

-- Step 2: Define an alias for the custom function
DEFINE DifferentialCalc com.example.pig.udf.DifferentialCalc();

-- Step 3: Execute the UDF inside a FOREACH projection
transformed_data = FOREACH sales GENERATE 
    index, 
    DifferentialCalc(units_sold, unit_cost) AS diff_val;

6.7.3 Student Questions and Answers

6.8 Apache Hive Data Warehousing Fundamentals & Architecture

6.8.1 Origins at Facebook and Data Warehousing Abstraction

Apache Hive was created at Facebook around 2006 to provide a data warehousing layer on top of Hadoop. At Facebook, data analysts and infrastructure teams managed petabyte-scale analytics datasets, but the vast majority of personnel were trained in SQL rather than Java MapReduce. The project was led by Jeff Hammerbacher's data team and became a top-level Apache project in 2010.

Hive introduces HiveQL (HQL), a declarative SQL-dialect that allows users to create tables, define schemas, and query HDFS datasets using familiar relational SQL queries.

Real-world: Hive is positioned as a Data Warehouse system. While Pig is used for procedural data transformation (ETL), Hive is used for structured data warehousing, schema enforcement, and analytical reporting across enterprise data lakes.

6.8.2 Hive Architectural Components: Driver, Compiler, and Execution Engine

The internal architecture of Apache Hive comprises four core services:

[ User Client / CLI / JDBC / ODBC ]
           │
           ▼
     [ Hive Driver ] ◄────────► [ Metastore (Derby / MySQL) ]
           │
           ▼
    [ Hive Compiler (Parser → Optimizer → Plan) ]
           │
           ▼
[ Execution Engine (MapReduce / Tez / Spark) ]
           │
           ▼
    [ Hadoop Cluster / HDFS ]

1. Driver: Manages the lifecycle of a HiveQL statement, receiving incoming queries from clients, maintaining session context, and coordinating with the Compiler and Execution Engine. Think of the Driver as the conductor of an orchestra — it does not play instruments but ensures every section comes in at the right time.

1. Compiler: Parses HiveQL syntax, performs semantic checks against table schemas stored in the Metastore, generates a logical execution plan, and optimizes the plan into a Directed Acyclic Graph (DAG) of MapReduce jobs. The compiler uses a cost-based optimizer (introduced in Hive 0.14) to choose the most efficient execution strategy.

1. Execution Engine: Submits generated MapReduce DAG tasks to YARN and monitors task execution until final query results are returned. Modern Hive supports three execution engines:

• MapReduce (mr) — the original, disk-based engine
• Tez — a DAG engine that avoids writing intermediate results to HDFS (faster)
• Spark — in-memory processing engine

1. Metastore: Stores table schema definitions, column data types, table locations in HDFS, and serialization/deserialization (SerDe) rules in a relational database. The metastore is the catalog that tells Hive where data lives and how to interpret it.

6.8.3 HiveQL to MapReduce Compilation DAG

SELECT region, SUM(revenue) AS total_revenue
FROM sales_db.sales_table
WHERE revenue > 10000.0
GROUP BY region
ORDER BY total_revenue DESC;

6.8.4 Student Questions and Answers

6.9 Hive Metastore Architectures

6.9.1 Metastore Concept: Separation of HDFS Data and RDBMS Metadata

A foundational architectural principle of Apache Hive is the strict separation of Data and Metadata:

• Data (Raw File Content): Stored in distributed HDFS files (e.g., CSV, ORC, Parquet files located at hdfs://namenode:9000/user/hive/warehouse/sales_db.db/sales_table). Hive never modifies the raw data files during queries — it reads them as-is.

• Metadata (Schema Definitions): Stored separately in a Relational Database Management System (RDBMS) called the Hive Metastore. Metadata includes table names, column names, column data types, table owner, partition keys, HDFS storage locations, and SerDe (Serializer/Deserializer) rules.

<!-- hive-site.xml — embedded Derby (default) -->
<property>
  <name>javax.jdo.option.ConnectionURL</name>
  <value>jdbc:derby:;databaseName=metastore_db;create=true</value>
</property>

<!-- hive-site.xml — remote MySQL metastore -->
<property>
  <name>javax.jdo.option.ConnectionURL</name>
  <value>jdbc:mysql://dbserver:3306/hive_metastore?createDatabaseIfNotExist=true</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionDriverName</name>
  <value>com.mysql.jdbc.Driver</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionUserName</name>
  <value>hive_user</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionPassword</name>
  <value>hive_password</value>
</property>

6.9.2 Embedded Metastore using Apache Derby

By default, Apache Hive includes an Embedded Metastore powered by Apache Derby, an embedded Java relational database.

Embedded Derby Characteristics

• Storage Location: Writes metadata files to a local disk directory (./metastore_db) on the client machine.
• Process Execution: Derby runs inside the same JVM process as the Hive CLI client.
• Critical Limitation: Derby supports only a single active connection at a time. If one user opens a Hive CLI session using Derby, no other user or client application can connect to Hive simultaneously. Attempting a second connection produces a lock error.

Primary Use Case: Embedded Derby is suitable strictly for local testing, initial learning, and single-user script development.

6.9.3 Remote and External Metastore using MySQL/RDBMS

To support enterprise multi-tenant production clusters, Hive must be configured with a Remote / External Metastore.

Remote MySQL Metastore Characteristics

• Storage Location: Metadata resides in a dedicated external multi-user RDBMS (such as MySQL, PostgreSQL, or Oracle).
• Network Access: Hive clients connect to a centralized Hive Metastore Service daemon via Thrift RPC network protocols.
• Multi-User Concurrency: Hundreds of concurrent users, BI tools (Tableau, PowerBI), and automated scripts can query Hive simultaneously without database lock conflicts.

6.9.4 Student Questions and Answers

6.10 Course Administration and Practical Setup Logistics

6.10.1 Cluster Setup Effort and Time Investment Expectations

Setting up enterprise big data software components locally requires substantial hands-on effort. Building a complete functional single-node pseudo-distributed Hadoop environment (HDFS, YARN, MapReduce, Sqoop, Flume, Pig, and Hive) involves extensive configuration file editing (core-site.xml, hdfs-site.xml, yarn-site.xml, hive-site.xml), environment variable pathing, and Java dependency debugging.

Pedagogical Expectations:

• Setting up core Hadoop (HDFS and YARN) for the first time requires approximately 8 hours of dedicated troubleshooting.
• Setting up Apache Pig requires approximately 3 additional hours.
• Setting up Apache Hive and configuring the metastore can require up to 16 hours of troubleshooting before achieving a fully functional setup.
• Overall Course Workload: As a 4-credit intensive course, students are expected to invest approximately 120 total hours across the semester (30 hours per credit) in self-study, hands-on lab environment setup, and script execution.

6.10.2 Assignment Structure: Group versus Individual Evaluation

In response to student feedback regarding past group assignment challenges across mixed student cohorts:

• Past Friction: Randomly assigned multi-cohort project groups experienced uneven student participation, communication barriers, and scheduling conflicts.
• Policy Flexibility: For upcoming major assignments, students will have the option to complete project deliverables either in self-selected collaborative groups or individually.

6.10.3 Scheduling Extra Live Hands-on Sessions

To reinforce theoretical concepts with practical terminal command walkthroughs, an extra hands-on demonstration class focusing on Hive setup, table creation, and HiveQL execution was scheduled based on student consensus for Saturday at 8:00 PM, followed by the regular Sunday morning session.

6.10.4 Student Questions and Answers

Exam Guidance Summary

• MapReduce Usability & Motivation: Be prepared to explain why Yahoo and Facebook developed Pig and Hive in 2004 to eliminate raw Java MapReduce boilerplate. Know the declarative (Hive) vs. procedural (Pig) distinction.

• Pig Execution Modes: Memorize launch flags and filesystem behaviors for Local Mode (pig -x local, local disk, single JVM) versus MapReduce Cluster Mode (pig, HDFS, YARN containers). Know when to use each.

• Pig Structural Limitations: Understand that Pig has no native storage engine, no background web server/daemon, no remote JDBC interface, and is unsuitable for sub-second real-time OLTP queries.

• Pig Complex Data Types: Be ready to write precise syntax representations for Tuples (1, 'Europe'), Bags { (1, 'Europe'), (2, 'Asia') }, and Maps ['key'#value]. Know the difference between a Tuple (single ordered record) and a Bag (unordered collection of Tuples).

• HDFS Output Directories: Remember that STORE relation INTO 'path' always targets directories, not single files, because parallel MapReduce reducers output separate part files (part-r-00000).

• Pig Operator Syntax: Know syntax and operational mechanics for LOAD, FILTER, FOREACH ... GENERATE, GROUP BY, ORDER BY, LIMIT, SPLIT, SAMPLE, UNION, and COGROUP.

• Top-N Non-Determinism Pitfall: Understand why calling LIMIT 10 without a preceding ORDER BY col DESC produces non-deterministic, arbitrary records rather than the true top 10. Always sort before limiting.

• Java UDF Integration: Know the 4-step workflow for custom Java UDFs: extend EvalFunc, override exec(), compile JAR, and execute REGISTER /path/to/jar and DEFINE.

• Hive Architectural Components: Be ready to sketch and explain the roles of Hive Driver, Compiler, Execution Engine, and Metastore. Know the data flow: Client → Driver → Compiler → Execution Engine → HDFS.

• Hive Metastore Distinctions: Be ready to compare Embedded Derby metastores (single connection limit, testing only) versus Remote MySQL metastores (multi-tenant concurrent access, production standard).

Key Industry Applications

• ETL Data Pipelines (Apache Pig):

Tech organizations (such as Yahoo and Twitter) used Pig Latin scripts to write complex Extract, Transform, Load (ETL) data pipelines that ingest web logs, strip corrupted records, compute derived fields, and output cleansed datasets. At its peak, over 80% of Yahoo's MapReduce jobs were generated by Pig Latin scripts. ETL remains a core use case in modern data engineering, though Spark has largely replaced Pig for new pipelines.

• Enterprise Data Warehousing (Apache Hive):

Organizations (such as Facebook) use Apache Hive as an enterprise data warehouse layer over petabyte-scale HDFS data lakes, allowing business analysts to run SQL queries and integrate SQL-based BI reporting tools (Tableau, PowerBI, Looker). Hive's SQL compatibility makes it the default query engine for data analysts who do not write Java or Python. Major cloud platforms (AWS EMR, Google Dataproc, Azure HDInsight) all offer managed Hive services.

• Custom Business Logic Execution (Pig UDFs):

Companies package proprietary algorithms (such as custom fraud detection scoring, domain-specific text parsers, or proprietary encryption routines) into custom Java UDFs registered inside Pig pipelines. UDFs allow Pig to handle business logic that goes beyond standard SQL operations — for example, parsing custom log formats or computing domain-specific risk scores.

• Multi-Tenant Enterprise Metastore Architecture:

Enterprise cloud data platforms deploy central remote MySQL/PostgreSQL Metastore services to allow hundreds of concurrent users, automated ETL jobs, and query engines (Hive, Spark SQL, Presto, Impala) to share unified table metadata across data lakes. The metastore has become a de facto standard catalog format — even non-Hive engines like Spark SQL and Presto read Hive metastore schemas to discover and query tables.