Apache Hive Bucketing Resolution, Relational Joins, Subquery Pipelines, Apache ZooKeeper, and Apache HBase

8.1 Hive Bucketing Data Types and Precision Resolution

8.1.1 Floating-Point Precision Defects in Bucketing Keys

In Apache Hive, bucketing organizes table data into discrete physical files based on the hash value of a designated bucketing column. Think of buckets as labeled drawers: every record gets a drawer number computed from its bucketing-column value. The hashing algorithm applies a modulo operation over the total number of defined buckets \(N\):

\[ \text{bucket\_id} = \text{hash}(\text{column\_value}) \pmod{N} \]

where:

• \(\text{hash}(\cdot)\) maps the column value to a large integer,
• \(N\) is the number of buckets defined in the table,
• the result is an index in the range \(0, 1, \dots, N-1\), meaning one of the \(N\) bucket files on HDFS.

For an integer column, the hash of the value is the value itself, so for an INT column the formula effectively becomes \(\text{bucket\_id} = \text{column\_value} \pmod{N}\). That is the clean case. The trouble starts when the bucketing column is a floating-point type.

A significant edge-case anomaly occurs when floating-point data types (FLOAT or DOUBLE) are selected as bucketing keys. In a previous session, a table named clustered_metals was created with a weight column specified as a FLOAT data type and partitioned into \(N = 3\) buckets.

When floating-point values undergo modulo arithmetic, IEEE 754 floating-point representations introduce internal binary rounding artifacts. For instance, an integer weight of \(3\) stored as a float may be represented internally as \(2.9999999\dots\). Evaluating \(2.9999999 \pmod 3\) produces a remainder near \(2.999\dots\), placing the record into bucket index 2 instead of bucket index 0.

During raw bucket file inspection, this precision artifact caused records with incompatible integer remainders to cluster into identical physical files. For example, records with weights \(2\) (\(2 \pmod 3 = 2\)) and \(6\) (\(6 \pmod 3 = 0\)) were improperly co-located inside bucket file 2, whereas weights \(4\) and \(7\) (\(4 \pmod 3 = 1\), \(7 \pmod 3 = 1\)) resided correctly in bucket file 1.

The fix: change the data type.

To resolve this precision defect, the table schema must be redefined to convert the bucketing column weight from FLOAT to INT:

CREATE TABLE clustered_metals (
    shop_id INT,
    metal STRING,
    short_form STRING,
    weight INT
)
CLUSTERED BY (weight) INTO 3 BUCKETS;

INSERT INTO TABLE clustered_metals 
SELECT shop_id, metal, short_form, weight 
FROM metals_temp;

After updating the column data type to INT and re-ingesting raw records from metals_temp, Hive distributes the records into three clean HDFS files inside the table directory.

8.1.2 Query Optimization and Full Table Scan Avoidance

Bucketing provides direct performance advantages by avoiding full table scans during filtered query execution. When a query contains a WHERE clause on a bucketed column, Hive calculates the target bucket index prior to reading data from storage.

Consider a query filtering records by weight:

SELECT * FROM clustered_metals WHERE weight = 1;

Instead of scanning all underlying HDFS data blocks, Hive evaluates \(1 \pmod 3 = 1\). The execution engine reads only bucket file 1 and completely skips bucket files 0 and 2.

Real-world: While bucketing 8 sample records across weights 1 through 8 shows negligible performance differences, bucketing becomes essential when processing multi-terabyte datasets containing high-cardinality columns (such as user IDs, phone numbers, or transaction timestamps). At enterprise scale, skipping unneeded bucket files reduces disk I/O and query execution time significantly. Enterprise data warehouses combine partitioning and bucketing over ORC/Parquet files to accelerate ad-hoc queries on petabyte-scale data lakes, skipping unneeded HDFS blocks during WHERE evaluation.

8.2 Relational Joins in Apache Hive

8.2.1 Relational Normalization and MapReduce Join Execution Mechanics

Database normalization decomposes flat tables into distinct entities to eliminate redundant data storage and prevent update anomalies. For example, maintaining employee names and salaries across separate salary and names tables prevents duplicating employee details when salary updates occur:

• salary table: id, salary_amount (e.g., 1 -> 10000, 2 -> 10500, 3 -> 15000)
• names table: id, employee_name (e.g., 1 -> John, 2 -> Clark, 3 -> Mark, 4 -> Alex)

When business logic requires a combined view of normalized tables, a relational join merges records across common keys. Under the hood, distributed join processing in MapReduce follows a two-step relational transformation:

Step 1 — Cartesian Cross Product. The execution engine pairs every record from Table A with every record from Table B. For a table with \(M\) rows and a table with \(N\) rows, the cross product generates \(M \times N\) intermediate tuples:

\[ |T_A \times T_B| = |T_A| \times |T_B| = 3 \times 4 = 12 \text{ intermediate tuples} \]

where \(|T_A|\) and \(|T_B|\) are the row counts of the two tables, and \(\times\) denotes the Cartesian product. The raw intermediate output pairs every salary entry with every name entry.

Step 2 — Predicate Verification and Filtering. The join operator evaluates the equality condition:

\[ \text{TableA.id} == \text{TableB.id} \]

Tuples failing this predicate match are discarded. Only tuples where TableA.id equals TableB.id are retained in the final output relation.

8.2.2 Hive HQL Join Syntax and Execution

In Hive QL, explicit table qualifications (table_name.column_name) resolve column ambiguities when joining multiple tables that share identical column identifiers.

Consider two normalized Hive tables:

• metals: Contains transaction records with columns shop_id, metal, short, and weight.
• metal_shops: Contains lookup metadata with columns shop_id and shop_name.

To project transaction data alongside shop names, the query must specify fully qualified column names in both the SELECT projection list and the ON join predicate:

SELECT metals.shop_id, metals.metal, metals.short, metal_shops.shop_name, metals.weight 
FROM metals 
JOIN metal_shops ON metals.shop_id = metal_shops.shop_id;

When this HQL query executes:

1. Hive compiles the declarative SQL statement into a MapReduce job plan and submits it to YARN.
2. The YARN ResourceManager allocates resources and launches an ApplicationMaster container on a cluster node.
3. Mappers tag input splits from metals and metal_shops with join key values (shop_id).
4. Reducers shuffle matching keys together, execute the join verification step, and output the joined dataset containing shop_id, metal, short, shop_name, and weight.

Real-world: Relational joins are the backbone of star-schema data warehouses, where a large fact table (transactions) joins to small dimension tables (shops, customers, dates) on key columns. Because Hive only supports equijoins, warehouse schemas are designed so that every join is an equality join on a key — a design rule that carries straight from Hive into modern SQL engines.

8.3 Hive Subqueries and Multi-Stage MapReduce Pipeline Execution

8.3.1 Subqueries and Derived Table Aggregations

Hive supports subqueries (also called inner queries) in the FROM clause. A subquery wraps an intermediate result set into a temporary named table alias, allowing outer queries to perform subsequent filtering or aggregation operations.

Suppose a business requirement calls for calculating the total weight of all metals purchased from each shop, aggregating across individual metal transactions. The inner query joins metals and metal_shops, and the outer query applies SUM(weight) grouped by shop_name:

SELECT shop_name, SUM(weight) 
FROM (
    SELECT metals.shop_id, metals.metal, metals.short, metal_shops.shop_name, metals.weight 
    FROM metals 
    JOIN metal_shops ON metals.shop_id = metal_shops.shop_id
) temp_table 
GROUP BY shop_name;

8.3.2 Multi-Stage MapReduce Execution Pipeline

A query that combines a join with an aggregation does not run as one MapReduce job. Hive breaks it into a pipeline of stages, where each stage is its own MapReduce job.

Inputs and outputs of the pipeline:

• Input: raw data splits from metals and metal_shops.
• Intermediate output (after Stage 1): the joined dataset — rows of shop_id, metal, short, shop_name, weight written to temporary HDFS files.
• Final output (after Stage 2): one row per distinct shop_name with its SUM(weight).

The two stages, step by step:

1. Stage 1 (MapReduce Job 1 — Relational Join):
   • Mapper 1: Reads raw data splits from metals and metal_shops, emitting tuples keyed by shop_id.
   • Reducer 1: Joins matching records on shop_id and writes the intermediate joined dataset to temporary HDFS files.
2. Stage 2 (MapReduce Job 2 — Group By Aggregation):
   • Mapper 2: Reads the intermediate join output generated by Reducer 1 and extracts shop_name as the grouping key and weight as the value.
   • Reducer 2: Computes SUM(weight) for each distinct shop_name key and writes final aggregated results to HDFS.

Real-world: The "join then aggregate" pattern is the shape of most reporting queries — total sales per region, total weight per shop, average spend per customer. Because each stage is a separate MapReduce job, production warehouses watch stage counts via EXPLAIN, and modern engines (Tez, Spark SQL) pipeline the same logic in memory to avoid repeated HDFS round-trips.

8.4 Distributed Coordination with Apache ZooKeeper

8.4.1 Coordination Challenges in Distributed Clusters

Distributed systems comprising active-passive master architectures and worker nodes encounter several critical operational challenges:

1. Active/Passive Master Coordination: In high-availability HDFS setups, a primary NameNode operates as the active master while a secondary NameNode remains passive. The passive node syncs state from fsimage and editlog files. If the active master crashes, an external coordinator must detect the failure, prevent split-brain scenarios (where both masters claim leadership), and promote the passive node to active master.
2. Task Handoff and State Recovery: In YARN, an ApplicationMaster manages container execution across slave nodes. If an ApplicationMaster crashes mid-job, the replacement node must recover the exact state of completed and pending tasks executed by its predecessor.
3. Dynamic Service Discovery and Centralized Configuration: Cluster services (such as Hive Metastore connections to MySQL databases) require centralized configuration management. When service endpoints change IP addresses dynamically across nodes, dependent services must discover the new endpoints automatically without hardcoded configuration edits.

8.4.2 ZooKeeper Ensemble Architecture, Quorum Math, and Consistency Guarantees

ZooKeeper operates as a small, highly redundant cluster of nodes called an ensemble.

+-------------------------------------------------------------+
|                      ZooKeeper Ensemble                     |
|                                                             |
|   +------------------+             +------------------+     |
|   |  Follower Node   |             |  Follower Node   |     |
|   +--------+---------+             +--------+---------+     |
|            ^                                ^               |
|            |    ZAB (Atomic Broadcast)      |               |
|            +---------------+----------------+               |
|                            |                                |
|                   +--------+---------+                      |
|                   |   Leader Node    |                      |
|                   +--------+---------+                      |
+----------------------------^--------------------------------+
                             | Write Requests
                    +--------+---------+
                    | ZooKeeper Client |
                    +------------------+

1. Leader-Follower Roles: One node in the ensemble is elected as the Leader, while all remaining nodes function as Followers. All write requests from application clients are directed to the Leader. Read requests can be served by any Leader or Follower node.
2. Quorum Mathematics and Odd Node Rule: An ensemble must contain an odd number of servers (\(n = 3, 5, 7\)) to ensure clear majority voting during leader elections and quorum approvals. The minimum quorum size \(Q\) required to confirm writes is:

\[ Q = \left\lfloor \frac{n}{2} \right\rfloor + 1 \]

where \(n\) is the total number of servers in the ensemble, \(\lfloor \cdot \rfloor\) is the floor function (round down to the nearest integer), and \(Q\) is the smallest integer strictly greater than half of \(n\) — a strict majority.

3. High Consistency via ZAB Protocol: ZooKeeper prioritizes consistency and partition tolerance (CP in the CAP theorem). Writes are replicated across ensemble nodes using the ZooKeeper Atomic Broadcast (ZAB) protocol, which runs in two phases:
   • Phase 1 — Leader election: the machines elect a distinguished Leader; the other machines (Followers) synchronize their state with it.
   • Phase 2 — Atomic broadcast: the Leader forwards every write request to the Followers and commits the change only when a majority have persisted it to disk.

The Leader acknowledges a write operation as successful to the client only after a quorum of nodes confirms writing the update to memory and disk. Subsequent reads served by any node return the updated consistent state. Because every write is committed to disk on a quorum before being acknowledged, a minority failure can never lose the latest state.

8.4.3 ZNode Types and Notification Watches

ZooKeeper structures stored data in a hierarchical tree format similar to a Unix file system. Each node in the tree is called a ZNode (ZooKeeper Node), identified by a slash-delimited path (such as /app/hadoop/master). Each ZNode can store small configuration datasets (typically under a few kilobytes).

                      / (Root)
                      |
        +-------------+-------------+
        |                           |
     /hadoop                     /services
        |                           |
     /master                     /hbase
  (ZNode: IP/Port)           (ZNode: Metadata)

ZooKeeper supports four distinct ZNode classifications:

1. Ephemeral ZNodes: Temporary ZNodes linked directly to a client session. When the client session terminates or misses periodic heartbeats, ZooKeeper automatically deletes the Ephemeral ZNode.
   • Use case: Dynamic failure detection. A master server creates an ephemeral ZNode at /hadoop/master. A ZooKeeper Fault Controller (zkfc) library embedded in each node sends heartbeats to ZooKeeper. If the active master crashes, its session expires, the ephemeral ZNode disappears, and ZooKeeper triggers an event to promote a standby master.
2. Persistent ZNodes: Permanent ZNodes that persist on disk regardless of client disconnects or node restarts.
   • Use case: Storing cluster configuration settings, active job progress metadata, and static service endpoints accessible to all nodes.
3. Container ZNodes: Specialized ZNodes used for distributed mutual exclusion and leader election. When a master node attempts to claim leadership, it creates a container ZNode. The first client to write succeeds and acquires an exclusive lock. Concurrent write requests from secondary clients are denied until the lock holder releases the ZNode or crashes.
4. Sequential ZNodes: ZNodes where ZooKeeper automatically appends a monotonically increasing 10-digit sequence number (such as node-0000000001) to the ZNode path, facilitating ordered queue management.

These ZNode types are only half the story. The other half is how clients learn that a ZNode has changed.

8.4.4 Modern Alternatives: Kubernetes and Choreographic Architecture

Real-world: While legacy platforms (such as Apache Storm, HBase, Solr, and early Apache Kafka versions) relied heavily on ZooKeeper for distributed coordination, modern cloud-native architectures are transitioning away from dedicated coordination clusters:

• Kafka KRaft Mode: Apache Kafka version 4.0+ completely removes the ZooKeeper dependency, replacing it with an event-driven internal consensus protocol called KRaft (Kafka Raft).
• Kubernetes and Choreographic Architecture: Modern microservice deployments favor decentralized choreographic patterns over centralized instructive controllers like ZooKeeper. In Kubernetes environments, containers contain self-managed health rules and auto-scaling definitions rather than depending on an external ZooKeeper ensemble.

Real-world: Apache ZooKeeper enables high-availability active/passive NameNode failover in HDFS, ResourceManager recovery in YARN, and dynamic service registration across distributed enterprise clusters — and it coordinates HBase's master election, which is exactly where we go next.

8.5 NoSQL Real-Time Distributed Storage with Apache HBase

8.5.1 Motivation, RDBMS Bottlenecks, and Anti-Patterns

Traditional Relational Database Management Systems (RDBMS) encounter severe scalability bottlenecks when handling massive volume and high-velocity throughput:

1. RDBMS Scale Bottlenecks: At petabyte scale with billions of rows (such as Facebook user profiles or Amazon shopping carts), relational joins fail to scale, index maintenance degrades write performance, disk I/O locks up, and manual sharding requires complex database administration.
2. Google Bigtable Paper: In 2006, Google published the foundational research paper Bigtable: A Distributed Storage System for Structured Data. Apache HBase was developed as an open-source implementation of Bigtable, running natively on top of HDFS.
3. Core Philosophy: HBase provides low-latency, real-time random read and write access to petabyte-scale data stored on HDFS. It forfeits complex SQL join capabilities and ACID transactions in favor of horizontally scalable key-value lookups (put and get).

When to Use HBase vs. Anti-Patterns.

8.5.2 HBase System Architecture

HBase follows a master-slave architecture built on top of HDFS and coordinated by Apache ZooKeeper:

+-----------------------------------------------------------------+
|                        HBase Architecture                       |
|                                                                 |
|                     +---------------------+                     |
|                     |   HMaster (Active)  |                     |
|                     +----------+----------+                     |
|                                | Coordinated via                |
|                                v ZooKeeper                      |
|                  +---------------------------+                  |
|                  | ZooKeeper Cluster / ZNode |                  |
|                  +-------------+-------------+                  |
|                                |                                |
|         +----------------------+----------------------+         |
|         v                                             v         |
| +---------------+                             +---------------+ |
| | RegionServer 1|                             | RegionServer 2| |
| | +-----------+ |                             | +-----------+ | |
| | | Region A  | |                             | | Region C  | | |
| | +-----------+ |                             | +-----------+ | |
| | | Region B  | |                             | | Region D  | | |
| | +-----------+ |                             | +-----------+ | |
| +-------+-------+                             +-------+-------+ |
|         |                                             |         |
+---------|---------------------------------------------|---------+
          v                                             v
+-----------------------------------------------------------------+
|               HDFS Storage Layer (DataNodes)                    |
+-----------------------------------------------------------------+

1. HMaster: Manages DDL operations (create, disable tables), monitors RegionServers, and performs region assignment. The HMaster handles initial client metadata requests. To avoid becoming a performance bottleneck, client read/write data operations bypass HMaster and communicate directly with RegionServers.
2. RegionServer: Manages and serves horizontal shards of table data called Regions. Each RegionServer handles data reads, writes, and local compaction operations across multiple regions.
3. ZooKeeper Coordination: Coordinates HMaster election (active/standby), maintains region lookup locations (meta table), and tracks active RegionServer health via ephemeral ZNodes.

8.5.3 HBase Multi-Dimensional Sparse Data Model

HBase organizes data into a 4-dimensional sparse mapping sorted lexicographically by Row Key:

\[ f(\text{RowKey}, \text{ColumnFamily}, \text{ColumnQualifier}, \text{Timestamp}) \rightarrow \text{CellValue} \]

where \(f\) maps the four coordinates to the value stored in the cell at their intersection.

Row Key: "2019MC04000"
  |-- Column Family: "user_info"
  |     |-- Column Qualifier: "name"  --> "Vignesh" (Timestamp: t1)
  |     |-- Column Qualifier: "photo" --> "http://site/pic.jpg" (Timestamp: t2)
  |-- Column Family: "payment_info"
        |-- Column Qualifier: "card"  --> "4111xxxx" (Timestamp: t3)

1. Row Key: A unique byte array identifier for every row. Rows are sorted lexicographically by Row Key. All data retrieval queries must specify a Row Key.
2. Column Family: A logical and physical grouping of related columns defined during table creation. Columns within the same Column Family are stored together on disk in dedicated storage files (HFiles).
   • Example: An e-commerce user table might define three Column Families: user_info (name, age), order_data (items, totals), and payment_info (card details).
3. Column Qualifier (Dynamic Columns): Dynamic attribute labels defined on the fly during record insertion under a specific Column Family (formatted as ColumnFamily:ColumnQualifier).
   • Sparsity: Rows do not need identical columns. If a row lacks a specific column, no storage space is consumed (unlike RDBMS null padding).
4. Timestamp / Multi-Version Concurrency Control (MVCC): Each cell maintains a configurable number of timestamped version histories. Inserting a new value into an existing cell creates a new version timestamped with system time.
   • Example: Cell revision history enables tracking historical password changes or Google Web Crawler page indexing revisions (com.cnn.com crawling history).
5. Data Type (ByteArray): HBase has only one native data type: byte[] (ByteArray). Strings, integers, floats, and binary objects are all stored as raw byte arrays. Applications handle serialization and type casting upon retrieval.

8.5.4 DDL and DML Data Operations (CLI and API)

HBase operations are executed through the HBase Shell (Ruby-based CLI), Java Client API, REST web gateway, Thrift (Facebook), or Avro.

Table Creation (DDL).

Creating an HBase table requires specifying only the table name and the Column Family names. Column qualifiers and data types are not defined at creation:

create 'user_table', 'user_info', 'payment_info'

Data Insertion (put - DML).

The put command requires the table name, target Row Key, target ColumnFamily:ColumnQualifier, and cell value:

put 'user_table', '2019MC04000', 'user_info:name', 'Vignesh'
put 'user_table', '2019MC04000', 'user_info:picture', 'http://store.com/vignesh.jpg'
put 'user_table', '2019MC04000', 'payment_info:card', '4111222233334444'

Data Retrieval (get - DML).

The get command retrieves data for a specific Row Key, with optional parameters specifying Column Families, dynamic qualifiers, or historical version counts:

get 'user_table', '2019MC04000'
get 'user_table', '2019MC04000', {COLUMNS => ['user_info:name'], VERSIONS => 3}

Real-world: Apache HBase powers real-time, high-concurrency low-latency applications including Facebook user profile lookups, Amazon shopping cart staging, Google search engine page-rank link indexing, and biometric authentication backends.

8.6 Exam Guidance Summary

8.6.1 Quiz Scope and Assignment Deadlines

Quiz 2 covers Apache Hive, ZooKeeper, HBase, NoSQL data management, and Apache Spark (scheduled immediately after completing the Spark module). The Hive, ZooKeeper, and HBase content from this lecture is therefore Quiz 2 material — keep the bucketing formula, the quorum formula, and the HBase data model fresh until then.

Assignment 1 will be released on Friday, September 10th, with a final submission deadline of October 15th. Students must implement a custom MapReduce join program in Java MapReduce and demonstrate identical query execution in Hive/Pig. The join mechanics from section 8.2 (mapper key emission, reducer cross-table matching) are exactly what this assignment asks you to build by hand.

8.6.2 Course Logistics and Class Scheduling

Mid-Semester regular exams occur on September 23rd, 24th, and 25th. Mid-Semester makeup exams occur on October 7th, 8th, and 9th. Plan the Assignment 1 submission (October 15th) against these dates — the assignment window overlaps the mid-semester exam period, so budget time accordingly.

8.7 Key Industry Applications

8.7.1 Enterprise Data Warehouse Query Acceleration

Real-world: Enterprise data warehouses use combined partitioning and bucketing over ORC/Parquet files to accelerate ad-hoc queries on petabyte-scale data lakes, skipping unneeded HDFS blocks during WHERE evaluation. The bucketing idea from section 8.1 — hash to a bucket, then read only the matching bucket file — is exactly the mechanism these warehouses exploit: coarse partitioning narrows the search to a directory, and bucketing narrows it further to a single file, so a WHERE on a high-cardinality key touches a tiny fraction of the data.

8.7.2 High-Availability Coordination and Real-Time NoSQL Storage

Real-world: Apache ZooKeeper enables high-availability active/passive NameNode failover in HDFS, ResourceManager recovery in YARN, and dynamic service registration across distributed enterprise clusters. Every place a cluster needs "one leader, no split-brain, and automatic failure detection," a ZooKeeper ensemble (or its modern successor) sits underneath — from Hadoop's own master nodes to Kafka's metadata management.

Real-world: Apache HBase powers real-time, high-concurrency low-latency applications including Facebook user profile lookups, Amazon shopping cart staging, Google search engine page-rank link indexing, and biometric authentication backends. These are all workloads that match the HBase use-case column from section 8.5: single-row key lookups on billions of rows, where a millisecond read by user ID beats any SQL feature set.