🧠 Hive Fundamentals

🎯 What is Apache Hive?

Apache Hive is a data warehouse software built on top of Hadoop. It provides an SQL-like interface (HiveQL) for querying and managing large datasets stored in HDFS. It translates SQL queries into MapReduce, Tez, or Spark jobs.

Why Hive Matters

SQL: Familiar interface for Big Data. Scale: Petabytes of data. Schema-on-read: Flexible. Batch: Analytics workloads. Ecosystem: Hadoop integration. Warehouse: Data warehousing.

📌 SQL on Hadoop📌 Data Warehouse📌 Schema-on-Read📌 Petabyte Scale
# Apache Hive — the big picture # Hive is SQL on Hadoop. # CONCEPTUAL OVERVIEW print(""" === Apache Hive Overview === Hive is: - Data warehouse on Hadoop - SQL-like interface (HiveQL) - Translates SQL → MapReduce/Tez/Spark - Schema-on-read - Petabyte scale Components: - HiveQL: SQL-like language - Metastore: Metadata repository - Driver: Query execution - Compiler: SQL → execution plan - Execution Engine: MapReduce/Tez/Spark Key features: - Tables (like RDBMS) - Partitions (for performance) - Buckets (for sampling/joins) - SerDe (serialization) - UDFs (custom functions) - ACID (transactions) """) # SIMPLE HIVEQL EXAMPLE print("=== HiveQL Example ===") print(""" -- Create database CREATE DATABASE IF NOT EXISTS analytics; -- Create table CREATE TABLE IF NOT EXISTS users ( id INT, name STRING, age INT, city STRING, salary DOUBLE ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE; -- Load data LOAD DATA LOCAL INPATH '/data/users.csv' INTO TABLE users; -- Query SELECT city, COUNT(*) as count, AVG(salary) as avg_salary FROM users WHERE age > 30 GROUP BY city ORDER BY count DESC; -- Output: -- NYC 2 62500.0 -- LA 1 55000.0 -- Chicago 1 80000.0 """) # KEY PROPERTIES print(""" === Key Properties of Hive === 1. SQL-LIKE INTERFACE - HiveQL (similar to SQL) - Familiar to analysts - Declarative queries 2. SCHEMA-ON-READ - Schema defined at query time - Flexible data evolution - Unlike RDBMS (schema-on-write) 3. BATCH PROCESSING - High latency (minutes) - Not for OLTP - For analytics 4. SCALABLE - Petabytes of data - Thousands of nodes - HDFS storage 5. EXTENSIBLE - UDFs (User Defined Functions) - SerDe (serialization) - Custom formats 6. METADATA - Metastore (derby/MySQL/PostgreSQL) - Table schemas - Partitions, buckets 7. ECOSYSTEM - HDFS storage - MapReduce/Tez/Spark execution - HBase integration """) # REAL-WORLD SCALE print("=== Real-World Hive Deployments ===") deployments = [ ('Facebook', '300+ PB', 'Largest Hive deployment'), ('Yahoo', '100+ PB', 'Search analytics'), ('Netflix', 'Petabytes', 'Recommendations'), ('LinkedIn', 'Petabytes', 'Analytics'), ('Twitter', 'Petabytes', 'Trends'), ('Airbnb', 'Petabytes', 'Search'), ] print(f"{'Company':<15} {'Data Volume':<20} {'Use Case':<30}") print("-" * 65) for company, volume, use_case in deployments: print(f"{company:<15} {volume:<20} {use_case:<30}") # HIVE USE CASES print("\n=== Hive Use Cases ===") print(""" 1. DATA WAREHOUSING - ETL pipelines - Historical analytics - Reporting 2. LOG ANALYSIS - Server logs - Application logs - Security logs 3. DATA MINING - Pattern discovery - Customer segmentation - Fraud detection 4. BUSINESS INTELLIGENCE - Dashboards - KPIs - Trend analysis 5. DATA PREPARATION - Cleaning - Transformation - Aggregation 6. MACHINE LEARNING - Feature engineering - Training data prep - Batch scoring """)

🏗️ Architecture

🎯 Hive Architecture

Hive's architecture consists of the HiveServer2 (query interface), Driver (query compilation), Metastore (metadata), and Execution Engine (MapReduce/Tez/Spark).

Architecture Components

HiveServer2: Query interface. Driver: Compiles queries. Compiler: SQL → plan. Metastore: Metadata. Execution: MapReduce/Tez/Spark. Storage: HDFS.

🏗️ HiveServer2🏗️ Driver🏗️ Metastore🏗️ Execution
# Hive Architecture print("=== Hive Architecture Diagram ===") print(""" ┌─────────────────────────────────────────┐ │ Client │ │ (Beeline, JDBC, ODBC, Thrift) │ └─────────────────┬───────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ HiveServer2 │ │ (Query Interface) │ └─────────────────┬───────────────────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ Driver │ │ │ │ ┌─────────────┐ ┌─────────────────┐ │ │ │ Compiler │ │ Optimizer │ │ │ │ │ │ │ │ │ │ SQL → Plan │ │ Rule-based │ │ │ └─────────────┘ └─────────────────┘ │ │ │ │ ┌─────────────┐ ┌─────────────────┐ │ │ │ Executor │ │ Metastore │ │ │ │ │ │ Client │ │ │ │ Run tasks │ │ │ │ │ └─────────────┘ └─────────────────┘ │ └─────────────────┬───────────────────────┘ │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │MapReduce│ │ Tez │ │ Spark │ │ │ │ │ │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ └───────────┼───────────┘ │ ▼ ┌─────────────────────────────────────────┐ │ HDFS │ │ (Data Storage) │ └─────────────────────────────────────────┘ """) # COMPONENTS print("\n=== Hive Components ===") print(""" 1. HIVESERVER2 - Query interface - JDBC/ODBC/Thrift - Multi-user - Authentication 2. DRIVER - Receives queries - Manages sessions - Coordinates execution 3. COMPILER - Parses HiveQL - Creates logical plan - Optimizes - Creates physical plan 4. METASTORE - Metadata repository - Table schemas - Partitions, buckets - Derby (default), MySQL, PostgreSQL 5. EXECUTION ENGINE - MapReduce (original) - Tez (DAG, faster) - Spark (in-memory) 6. STORAGE - HDFS - HBase - S3 - Other Hadoop-compatible """) # QUERY LIFECYCLE print("\n=== Query Lifecycle ===") print(""" 1. SUBMIT - Client submits HiveQL - HiveServer2 receives 2. PARSE - Parse query - Validate syntax 3. COMPILE - Semantic analysis - Logical plan - Optimization 4. OPTIMIZE - Predicate pushdown - Join reordering - Partition pruning 5. PLAN - Physical plan - MapReduce/Tez/Spark jobs 6. EXECUTE - Submit jobs - Monitor progress 7. RETURN - Fetch results - Return to client """) # IMPLEMENTATION print("\n=== Simulating Hive ===") class HiveMetastore: """Simulate Hive Metastore.""" def __init__(self): self.databases = {'default': {}} self.current_db = 'default' def create_database(self, name): if name not in self.databases: self.databases[name] = {} print(f"Metastore: Created database '{name}'") def use_database(self, name): if name in self.databases: self.current_db = name print(f"Metastore: Using database '{name}'") def create_table(self, name, schema, partition_cols=None): table = { 'name': name, 'schema': schema, 'partition_cols': partition_cols or [], 'partitions': [], 'location': f"/user/hive/warehouse/{name}", } self.databases[self.current_db][name] = table print(f"Metastore: Created table '{name}'") return table def get_table(self, name): return self.databases[self.current_db].get(name) def add_partition(self, table_name, partition_values): table = self.get_table(table_name) if table: table['partitions'].append(partition_values) print(f"Metastore: Added partition {partition_values} to '{table_name}'") def list_tables(self): return list(self.databases[self.current_db].keys()) class HiveDriver: """Simulate Hive Driver.""" def __init__(self, metastore): self.metastore = metastore self.queries = [] def execute(self, query): """Execute HiveQL query.""" self.queries.append(query) print(f"\nDriver: Executing query") print(f" {query[:80]}...") # Parse query query_lower = query.lower().strip() if query_lower.startswith('create database'): db_name = query.split()[2] self.metastore.create_database(db_name) return "Database created" elif query_lower.startswith('use'): db_name = query.split()[1] self.metastore.use_database(db_name) return f"Using {db_name}" elif query_lower.startswith('create table'): # Extract table name parts = query.split() table_name = parts[2] # Parse schema (simplified) schema_start = query.find('(') schema_end = query.find(')') schema_str = query[schema_start+1:schema_end] schema = [col.strip() for col in schema_str.split(',')] self.metastore.create_table(table_name, schema) return f"Table {table_name} created" elif query_lower.startswith('select'): return self._execute_select(query) else: return "Query executed" def _execute_select(self, query): """Execute SELECT query.""" # Simulate result return [ {'city': 'NYC', 'count': 2, 'avg_salary': 62500.0}, {'city': 'LA', 'count': 1, 'avg_salary': 55000.0}, {'city': 'Chicago', 'count': 1, 'avg_salary': 80000.0}, ] def explain(self, query): """Show query plan.""" print("\n=== Query Plan ===") print("Stage 1: TableScan (users)") print(" → Filter (age > 30)") print(" → GroupBy (city)") print(" → Aggregate (count, avg)") print("Stage 2: Reduce") print(" → Sort (count DESC)") print(" → Output") # Simulate metastore = HiveMetastore() driver = HiveDriver(metastore) print("=== Hive Simulation ===") # Create database driver.execute("CREATE DATABASE analytics") driver.execute("USE analytics") # Create table driver.execute(""" CREATE TABLE users ( id INT, name STRING, age INT, city STRING, salary DOUBLE ) """) # Add partition metastore.add_partition('users', ['2024', '01']) # Query result = driver.execute(""" SELECT city, COUNT(*) as count, AVG(salary) as avg_salary FROM users WHERE age > 30 GROUP BY city ORDER BY count DESC """) print("\nQuery Result:") for row in result: print(f" {row}") # Explain driver.explain("SELECT * FROM users WHERE age > 30") # List tables print(f"\nTables in 'analytics': {metastore.list_tables()}") # ARCHITECTURE SUMMARY print("\n=== Architecture Summary ===") print(f"{'Component':<25} {'Role':<40}") print("-" * 65) print(f"{'HiveServer2':<25} {'Query interface':<40}") print(f"{'Driver':<25} {'Query compilation':<40}") print(f"{'Compiler':<25} {'SQL to plan':<40}") print(f"{'Metastore':<25} {'Metadata repository':<40}") print(f"{'Execution Engine':<25} {'MapReduce/Tez/Spark':<40}") print(f"{'HDFS':<25} {'Data storage':<40}")

🗄️ Metastore

🎯 The Metadata Repository

The Metastore stores all metadata about Hive tables: schemas, partitions, columns, data types, and locations. It's the central catalog of the data warehouse.

Metastore

Metadata: Table schemas. Partitions: Partition info. Storage: Derby/MySQL/PostgreSQL. Thrift: API. Catalog: Central registry. Integration: Spark, Impala, Presto.

🗄️ Metadata🗄️ Schema🗄️ Partitions🗄️ Catalog
# Hive Metastore print("=== What is the Metastore? ===") print(""" Metastore: - Central metadata repository - Stores table definitions - Stores partitions - Stores columns and types - Stores locations - Stores SerDe info Without metastore: - Hive doesn't know about tables - No schema information - No partitions """) print("\n=== Metastore Contents ===") print(""" 1. DATABASES - Database names - Locations - Properties 2. TABLES - Table names - Columns and types - Storage format - Location - SerDe 3. PARTITIONS - Partition keys - Partition values - Locations 4. COLUMNS - Column names - Data types - Comments 5. INDEXES - Index definitions - Index locations 6. ROLES & PRIVILEGES - Security - Access control 7. STATISTICS - Row counts - Size - Column stats """) print("\n=== Metastore Configuration ===") print(""" Databases: 1. Derby (default) - Embedded - Single user - Not for production 2. MySQL (recommended) - Multi-user - Production - Scalable 3. PostgreSQL - Alternative - Production - Scalable Configuration: javax.jdo.option.ConnectionURL=jdbc:mysql://localhost/metastore javax.jdo.option.ConnectionDriverName=com.mysql.jdbc.Driver javax.jdo.option.ConnectionUserName=hive javax.jdo.option.ConnectionPassword=password """) print("\n=== Metastore Modes ===") print(""" 1. EMBEDDED METASTORE - Derby database - Same JVM as Hive - Single user - Testing only 2. LOCAL METASTORE - External database (MySQL) - Same JVM as Hive - Multiple users (same machine) - Better than embedded 3. REMOTE METASTORE - External database - Separate service (Thrift) - Multiple users - Production """) # IMPLEMENTATION print("\n=== Simulating Metastore ===") import sqlite3 import json class MetastoreDB: """Simulate Hive Metastore database.""" def __init__(self, db_path=':memory:'): self.conn = sqlite3.connect(db_path) self.cursor = self.conn.cursor() self._init_schema() def _init_schema(self): """Create metastore tables.""" self.cursor.executescript(''' CREATE TABLE IF NOT EXISTS DBS ( DB_ID INTEGER PRIMARY KEY, NAME TEXT UNIQUE, LOCATION TEXT ); CREATE TABLE IF NOT EXISTS TBLS ( TBL_ID INTEGER PRIMARY KEY, DB_ID INTEGER, TBL_NAME TEXT, TBL_TYPE TEXT, LOCATION TEXT, FOREIGN KEY (DB_ID) REFERENCES DBS(DB_ID) ); CREATE TABLE IF NOT EXISTS COLUMNS ( COLUMN_ID INTEGER PRIMARY KEY, TBL_ID INTEGER, COLUMN_NAME TEXT, TYPE_NAME TEXT, FOREIGN KEY (TBL_ID) REFERENCES TBLS(TBL_ID) ); CREATE TABLE IF NOT EXISTS PARTITIONS ( PART_ID INTEGER PRIMARY KEY, TBL_ID INTEGER, PART_NAME TEXT, LOCATION TEXT, FOREIGN KEY (TBL_ID) REFERENCES TBLS(TBL_ID) ); ''') self.conn.commit() # Create default database self.cursor.execute( "INSERT OR IGNORE INTO DBS (DB_ID, NAME, LOCATION) VALUES (1, 'default', '/user/hive/warehouse')" ) self.conn.commit() def create_database(self, name, location=None): """Create a database.""" loc = location or f"/user/hive/warehouse/{name}.db" self.cursor.execute( "INSERT OR IGNORE INTO DBS (NAME, LOCATION) VALUES (?, ?)", (name, loc) ) self.conn.commit() print(f"Metastore: Created database '{name}'") def get_database_id(self, name): """Get database ID.""" self.cursor.execute("SELECT DB_ID FROM DBS WHERE NAME = ?", (name,)) result = self.cursor.fetchone() return result[0] if result else None def create_table(self, db_name, table_name, columns, location=None): """Create a table.""" db_id = self.get_database_id(db_name) if not db_id: print(f"Database '{db_name}' not found") return None loc = location or f"/user/hive/warehouse/{table_name}" self.cursor.execute( "INSERT INTO TBLS (DB_ID, TBL_NAME, TBL_TYPE, LOCATION) VALUES (?, ?, ?, ?)", (db_id, table_name, 'MANAGED_TABLE', loc) ) tbl_id = self.cursor.lastrowid for col_name, col_type in columns: self.cursor.execute( "INSERT INTO COLUMNS (TBL_ID, COLUMN_NAME, TYPE_NAME) VALUES (?, ?, ?)", (tbl_id, col_name, col_type) ) self.conn.commit() print(f"Metastore: Created table '{db_name}.{table_name}' with {len(columns)} columns") return tbl_id def get_table(self, db_name, table_name): """Get table info.""" db_id = self.get_database_id(db_name) if not db_id: return None self.cursor.execute( "SELECT TBL_ID, LOCATION FROM TBLS WHERE DB_ID = ? AND TBL_NAME = ?", (db_id, table_name) ) result = self.cursor.fetchone() if not result: return None tbl_id, location = result self.cursor.execute( "SELECT COLUMN_NAME, TYPE_NAME FROM COLUMNS WHERE TBL_ID = ?", (tbl_id,) ) columns = self.cursor.fetchall() return { 'database': db_name, 'table': table_name, 'location': location, 'columns': columns, } def add_partition(self, db_name, table_name, partition_name, location=None): """Add a partition.""" db_id = self.get_database_id(db_name) self.cursor.execute( "SELECT TBL_ID FROM TBLS WHERE DB_ID = ? AND TBL_NAME = ?", (db_id, table_name) ) result = self.cursor.fetchone() if not result: print(f"Table '{table_name}' not found") return tbl_id = result[0] loc = location or f"/user/hive/warehouse/{table_name}/{partition_name}" self.cursor.execute( "INSERT INTO PARTITIONS (TBL_ID, PART_NAME, LOCATION) VALUES (?, ?, ?)", (tbl_id, partition_name, loc) ) self.conn.commit() print(f"Metastore: Added partition '{partition_name}' to '{table_name}'") def list_databases(self): """List databases.""" self.cursor.execute("SELECT NAME FROM DBS") return [r[0] for r in self.cursor.fetchall()] def list_tables(self, db_name): """List tables in database.""" db_id = self.get_database_id(db_name) if not db_id: return [] self.cursor.execute("SELECT TBL_NAME FROM TBLS WHERE DB_ID = ?", (db_id,)) return [r[0] for r in self.cursor.fetchall()] def list_partitions(self, db_name, table_name): """List partitions for table.""" db_id = self.get_database_id(db_name) self.cursor.execute( "SELECT TBL_ID FROM TBLS WHERE DB_ID = ? AND TBL_NAME = ?", (db_id, table_name) ) result = self.cursor.fetchone() if not result: return [] tbl_id = result[0] self.cursor.execute( "SELECT PART_NAME FROM PARTITIONS WHERE TBL_ID = ?", (tbl_id,) ) return [r[0] for r in self.cursor.fetchall()] # Demo print("\n=== Metastore Demo ===") ms = MetastoreDB() # Create databases ms.create_database('analytics') ms.create_database('staging') print(f"\nDatabases: {ms.list_databases()}") # Create table ms.create_table('analytics', 'users', [ ('id', 'int'), ('name', 'string'), ('age', 'int'), ('city', 'string'), ('salary', 'double'), ]) # Create partitioned table ms.create_table('analytics', 'events', [ ('event_id', 'int'), ('user_id', 'int'), ('event_type', 'string'), ('timestamp', 'bigint'), ]) # Add partitions ms.add_partition('analytics', 'events', 'year=2024/month=01/day=01') ms.add_partition('analytics', 'events', 'year=2024/month=01/day=02') ms.add_partition('analytics', 'events', 'year=2024/month=01/day=03') # Query metastore print("\n=== Query Metastore ===") print(f"Tables in analytics: {ms.list_tables('analytics')}") print(f"Partitions in events: {ms.list_partitions('analytics', 'events')}") # Get table details table = ms.get_table('analytics', 'users') print(f"\nTable details:") print(f" Database: {table['database']}") print(f" Table: {table['table']}") print(f" Location: {table['location']}") print(f" Columns:") for col, typ in table['columns']: print(f" {col}: {typ}") # METASTORE SUMMARY print("\n=== Metastore Summary ===") print(f"{'Aspect':<25} {'Description':<40}") print("-" * 65) print(f"{'Purpose':<25} {'Metadata repository':<40}") print(f"{'Contents':<25} {'Databases, tables, columns':<40}") print(f"{'Storage':<25} {'Derby, MySQL, PostgreSQL':<40}") print(f"{'API':<25} {'Thrift':<40}") print(f"{'Integration':<25} {'Spark, Impala, Presto':<40}") print(f"{'Modes':<25} {'Embedded, local, remote':<40}")

📝 HiveQL

🎯 SQL-like Query Language

HiveQL is Hive's SQL-like query language. It supports most SQL operations: SELECT, JOIN, GROUP BY, and more, with extensions for Big Data.

HiveQL

DDL: CREATE, DROP, ALTER. DML: INSERT, LOAD. Queries: SELECT, JOIN. Aggregates: COUNT, SUM, AVG. Window: OVER, PARTITION. Extensions: Lateral view, explode.

📝 DDL📝 DML📝 Queries📝 Window
# HiveQL — SQL on Hadoop print("=== HiveQL Overview ===") print(""" HiveQL is similar to SQL but with extensions: Common operations: - DDL: CREATE, DROP, ALTER - DML: INSERT, LOAD, UPDATE, DELETE - Queries: SELECT, JOIN, GROUP BY, ORDER BY - Aggregates: COUNT, SUM, AVG, MIN, MAX - Window functions: OVER, PARTITION BY - Extensions: LATERAL VIEW, EXPLODE, MAP Differences from SQL: - Schema-on-read - Batch processing (high latency) - No transactions (until ACID) - HDFS storage """) print("\n=== DDL (Data Definition Language) ===") print(""" -- Create database CREATE DATABASE IF NOT EXISTS analytics COMMENT 'Analytics database' LOCATION '/user/hive/warehouse/analytics.db'; -- Use database USE analytics; -- Create table CREATE TABLE IF NOT EXISTS users ( id INT COMMENT 'User ID', name STRING COMMENT 'User name', age INT COMMENT 'Age', city STRING COMMENT 'City', salary DOUBLE COMMENT 'Annual salary' ) COMMENT 'User information' ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\\n' STORED AS TEXTFILE; -- Create partitioned table CREATE TABLE IF NOT EXISTS events ( event_id INT, user_id INT, event_type STRING, timestamp BIGINT ) PARTITIONED BY (year INT, month INT, day INT) STORED AS PARQUET; -- Create external table CREATE EXTERNAL TABLE IF NOT EXISTS logs ( timestamp STRING, level STRING, message STRING ) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t' LOCATION '/data/logs'; -- Alter table ALTER TABLE users ADD COLUMNS (email STRING); ALTER TABLE users RENAME TO customers; -- Drop table DROP TABLE IF EXISTS users; -- Drop database DROP DATABASE IF EXISTS analytics CASCADE; """) print("\n=== DML (Data Manipulation Language) ===") print(""" -- Load data from local LOAD DATA LOCAL INPATH '/data/users.csv' INTO TABLE users; -- Load data from HDFS LOAD DATA INPATH '/data/users.csv' INTO TABLE users; -- Load partitioned data LOAD DATA LOCAL INPATH '/data/events_2024_01_01.csv' INTO TABLE events PARTITION (year=2024, month=1, day=1); -- Insert from query INSERT INTO TABLE users SELECT id, name, age, city, salary FROM staging_users WHERE age > 18; -- Insert overwrite INSERT OVERWRITE TABLE users SELECT * FROM staging_users; -- Dynamic partition insert INSERT INTO TABLE events PARTITION (year, month, day) SELECT event_id, user_id, event_type, timestamp, year, month, day FROM raw_events; -- Update (ACID only) UPDATE users SET salary = salary * 1.1 WHERE city = 'NYC'; -- Delete (ACID only) DELETE FROM users WHERE age < 18; """) print("\n=== Queries ===") print(""" -- Basic select SELECT * FROM users LIMIT 10; -- Filter SELECT name, age FROM users WHERE age > 30; -- Group by SELECT city, COUNT(*) as count, AVG(salary) as avg_salary FROM users GROUP BY city; -- Having SELECT city, AVG(salary) as avg_salary FROM users GROUP BY city HAVING avg_salary > 60000; -- Order by SELECT * FROM users ORDER BY salary DESC LIMIT 10; -- Join SELECT u.name, o.order_id, o.amount FROM users u JOIN orders o ON u.id = o.user_id; -- Left join SELECT u.name, o.order_id FROM users u LEFT JOIN orders o ON u.id = o.user_id; -- Subquery SELECT * FROM users WHERE salary > (SELECT AVG(salary) FROM users); -- Union SELECT name FROM users UNION ALL SELECT name FROM customers; -- Window functions SELECT name, city, salary, ROW_NUMBER() OVER (PARTITION BY city ORDER BY salary DESC) as rank FROM users; """) print("\n=== Advanced HiveQL ===") print(""" -- Lateral view with explode SELECT name, skill FROM users LATERAL VIEW EXPLODE(skills) skills_table AS skill; -- Map and reduce SELECT TRANSFORM(name, age) USING 'python3 script.py' AS (new_name, new_age) FROM users; -- CTE (Common Table Expression) WITH high_earners AS ( SELECT * FROM users WHERE salary > 100000 ) SELECT city, COUNT(*) FROM high_earners GROUP BY city; -- Analytical functions SELECT name, salary, SUM(salary) OVER (ORDER BY id) as running_total, AVG(salary) OVER (PARTITION BY city) as city_avg FROM users; """) # IMPLEMENTATION print("\n=== Simulating HiveQL ===") class HiveQLSimulator: """Simulate HiveQL execution.""" def __init__(self): self.data = { 'users': [ {'id': 1, 'name': 'Alice', 'age': 30, 'city': 'NYC', 'salary': 50000}, {'id': 2, 'name': 'Bob', 'age': 25, 'city': 'LA', 'salary': 60000}, {'id': 3, 'name': 'Charlie', 'age': 35, 'city': 'NYC', 'salary': 75000}, {'id': 4, 'name': 'Diana', 'age': 28, 'city': 'LA', 'salary': 55000}, {'id': 5, 'name': 'Eve', 'age': 32, 'city': 'Chicago', 'salary': 80000}, ], 'orders': [ {'order_id': 101, 'user_id': 1, 'amount': 100}, {'order_id': 102, 'user_id': 1, 'amount': 200}, {'order_id': 103, 'user_id': 2, 'amount': 150}, {'order_id': 104, 'user_id': 3, 'amount': 300}, {'order_id': 105, 'user_id': 3, 'amount': 250}, ], } def select(self, table, columns='*', where=None, group_by=None, aggregates=None, order_by=None, limit=None): """Execute SELECT query.""" data = self.data.get(table, []) # WHERE if where: col, op, val = where if op == '>': data = [r for r in data if r.get(col, 0) > val] elif op == '<': data = [r for r in data if r.get(col, 0) < val] elif op == '=': data = [r for r in data if r.get(col) == val] # GROUP BY if group_by: from collections import defaultdict grouped = defaultdict(list) for row in data: grouped[row[group_by]].append(row) result = [] for key, rows in grouped.items(): row_result = {group_by: key} if aggregates: for agg_name, agg_func, agg_col in aggregates: values = [r[agg_col] for r in rows] if agg_func == 'COUNT': row_result[agg_name] = len(values) elif agg_func == 'SUM': row_result[agg_name] = sum(values) elif agg_func == 'AVG': row_result[agg_name] = sum(values) / len(values) elif agg_func == 'MIN': row_result[agg_name] = min(values) elif agg_func == 'MAX': row_result[agg_name] = max(values) result.append(row_result) data = result # ORDER BY if order_by: col, asc = order_by data = sorted(data, key=lambda r: r.get(col, 0), reverse=not asc) # LIMIT if limit: data = data[:limit] # SELECT columns if columns != '*': col_list = [c.strip() for c in columns.split(',')] data = [{k: r.get(k) for k in col_list} for r in data] return data def join(self, left_table, right_table, left_key, right_key): """Execute JOIN.""" left_data = self.data.get(left_table, []) right_data = self.data.get(right_table, []) result = [] for l in left_data: for r in right_data: if l[left_key] == r[right_key]: merged = {**l, **r} result.append(merged) return result # Demo print("\n=== HiveQL Demo ===") hive = HiveQLSimulator() print("1. Basic SELECT:") result = hive.select('users', columns='name, age, city') for row in result: print(f" {row}") print("\n2. WHERE (age > 30):") result = hive.select('users', where=('age', '>', 30)) for row in result: print(f" {row['name']}: {row['age']}") print("\n3. GROUP BY city with aggregates:") result = hive.select( 'users', group_by='city', aggregates=[ ('count', 'COUNT', 'id'), ('avg_salary', 'AVG', 'salary'), ] ) for row in result: print(f" {row}") print("\n4. ORDER BY salary DESC:") result = hive.select('users', order_by=('salary', False), limit=3) for row in result: print(f" {row['name']}: {row['salary']}") print("\n5. JOIN users + orders:") result = hive.join('users', 'orders', 'id', 'user_id') for row in result: print(f" {row['name']} → order {row['order_id']}: ${row['amount']}") # HIVEQL SUMMARY print("\n=== HiveQL Summary ===") print(f"{'Category':<20} {'Commands':<45}") print("-" * 70) print(f"{'DDL':<20} {'CREATE, DROP, ALTER':<45}") print(f"{'DML':<20} {'INSERT, LOAD, UPDATE, DELETE':<45}") print(f"{'Queries':<20} {'SELECT, JOIN, GROUP BY, ORDER BY':<45}") print(f"{'Aggregates':<20} {'COUNT, SUM, AVG, MIN, MAX':<45}") print(f"{'Window':<20} {'OVER, PARTITION BY, ROW_NUMBER':<45}") print(f"{'Extensions':<20} {'LATERAL VIEW, EXPLODE, TRANSFORM':<45}")

📊 Tables & Partitions

🎯 Organizing Data

Hive tables organize data into rows and columns. Partitions divide tables into smaller parts based on column values, improving query performance by pruning irrelevant data.

Tables & Partitions

Managed: Hive manages data. External: Hive manages metadata. Partition: Directory split. Static: Manual partition. Dynamic: Auto partition. Pruning: Skip partitions.

📊 Managed📊 External📊 Partition📊 Pruning
# Hive Tables and Partitions print("=== Table Types ===") print(""" 1. MANAGED TABLE (Internal) - Hive manages data AND metadata - Data stored in Hive warehouse - DROP TABLE deletes data - Default table type CREATE TABLE users (...); 2. EXTERNAL TABLE - Hive manages metadata only - Data stored externally - DROP TABLE deletes metadata only - Data remains CREATE EXTERNAL TABLE logs (...) LOCATION '/data/logs'; 3. TEMPORARY TABLE - Session-scoped - Automatically dropped - Not stored in metastore CREATE TEMPORARY TABLE temp_users (...); """) print("\n=== Managed vs External ===") print(f"{'Aspect':<25} {'Managed':<25} {'External':<25}") print("-" * 75) print(f"{'Data location':<25} {'Hive warehouse':<25} {'External':<25}") print(f"{'DROP TABLE':<25} {'Deletes data':<25} {'Keeps data':<25}") print(f"{'Schema change':<25} {'Hive manages':<25} {'User manages':<25}") print(f"{'Use case':<25} {'ETL, intermediate':<25} {'Raw data, shared':<25}") print(f"{'Safety':<25} {'Lower':<25} {'Higher':<25}") print("\n=== Partitioning ===") print(""" Partitioning: - Divides table into parts - Based on column values - Creates subdirectories - Improves query performance Example: Table: events Partition by: year, month, day HDFS structure: /user/hive/warehouse/events/year=2024/month=01/day=01/ /user/hive/warehouse/events/year=2024/month=01/day=02/ /user/hive/warehouse/events/year=2024/month=02/day=01/ ... Benefits: - Partition pruning: Skip irrelevant partitions - Faster queries - Easier data management Cost: - Too many partitions: Metadata overhead - Small files: HDFS inefficiency - Recommend: 100s-1000s of partitions """) print("\n=== Static vs Dynamic Partitioning ===") print(""" STATIC PARTITIONING: - Partition values specified manually - Explicit INSERT - Slower to load - More control INSERT INTO events PARTITION (year=2024, month=1, day=1) SELECT event_id, user_id, event_type, timestamp FROM raw_events WHERE year=2024 AND month=1 AND day=1; DYNAMIC PARTITIONING: - Partition values from data - Automatic - Faster to load - Less control INSERT INTO events PARTITION (year, month, day) SELECT event_id, user_id, event_type, timestamp, year, month, day FROM raw_events; Configuration: hive.exec.dynamic.partition=true hive.exec.dynamic.partition.mode=nonstrict """) # IMPLEMENTATION print("\n=== Simulating Partitions ===") class PartitionedTable: """Simulate Hive partitioned table.""" def __init__(self, name, columns, partition_cols): self.name = name self.columns = columns self.partition_cols = partition_cols self.partitions = {} # (partition_values) -> data self.data = [] # Non-partitioned data def add_partition(self, partition_values, data): """Add data to a partition.""" key = tuple(partition_values) self.partitions[key] = data print(f" Added {len(data)} rows to partition {partition_values}") def query(self, where=None, partition_filter=None): """Query with partition pruning.""" print(f"\nQuerying {self.name}:") if partition_filter: print(f" Partition filter: {partition_filter}") # Partition pruning matching = [] for part_key, part_data in self.partitions.items(): if self._matches_filter(part_key, partition_filter): matching.extend(part_data) print(f" Scanning partition {part_key}") else: print(f" Pruning partition {part_key}") data = matching else: data = [] for part_data in self.partitions.values(): data.extend(part_data) # Apply WHERE if where: col, op, val = where if op == '>': data = [r for r in data if r.get(col, 0) > val] elif op == '<': data = [r for r in data if r.get(col, 0) < val] elif op == '=': data = [r for r in data if r.get(col) == val] return data def _matches_filter(self, partition_values, filter_dict): """Check if partition matches filter.""" for i, col in enumerate(self.partition_cols): if col in filter_dict: if partition_values[i] != filter_dict[col]: return False return True def list_partitions(self): return list(self.partitions.keys()) # Demo print("=== Partitioned Table Demo ===") # Create partitioned table events = PartitionedTable( 'events', ['event_id', 'user_id', 'event_type', 'timestamp'], ['year', 'month', 'day'] ) # Add partitions print("\nAdding partitions:") events.add_partition([2024, 1, 1], [ {'event_id': 1, 'user_id': 101, 'event_type': 'click', 'timestamp': 1704067200}, {'event_id': 2, 'user_id': 102, 'event_type': 'view', 'timestamp': 1704067300}, ]) events.add_partition([2024, 1, 2], [ {'event_id': 3, 'user_id': 101, 'event_type': 'purchase', 'timestamp': 1704153600}, ]) events.add_partition([2024, 2, 1], [ {'event_id': 4, 'user_id': 103, 'event_type': 'click', 'timestamp': 1706745600}, ]) print(f"\nPartitions: {events.list_partitions()}") # Query with partition pruning print("\n=== Query with Partition Pruning ===") result = events.query(partition_filter={'year': 2024, 'month': 1}) print(f"Result: {len(result)} rows") # Query with WHERE print("\n=== Query with WHERE ===") result = events.query(where=('user_id', '=', 101)) print(f"Result: {len(result)} rows") for row in result: print(f" {row}") # PARTITIONING BEST PRACTICES print("\n=== Partitioning Best Practices ===") print(""" 1. CHOOSE GOOD PARTITION COLUMNS - Low cardinality (few distinct values) - Commonly filtered - Time-based (year, month, day) 2. AVOID OVER-PARTITIONING - Too many small files - Metadata overhead - Aim for 100s-1000s partitions 3. USE DYNAMIC PARTITIONING - For ETL pipelines - Configure carefully 4. PARTITION PRUNING - Filter on partition columns - Hive skips irrelevant partitions 5. MULTI-LEVEL PARTITIONING - year/month/day - country/city - Don't exceed 3 levels 6. COMPACTION - Merge small files - Periodically - Use ALTER TABLE ... CONCATENATE """) # TABLE SUMMARY print("\n=== Table Summary ===") print(f"{'Type':<25} {'Data Location':<25} {'DROP Behavior':<25}") print("-" * 75) print(f"{'Managed':<25} {'Hive warehouse':<25} {'Deletes data':<25}") print(f"{'External':<25} {'External location':<25} {'Keeps data':<25}") print(f"{'Temporary':<25} {'Session':<25} {'Auto-dropped':<25}") print(f"{'Partitioned':<25} {'Subdirectories':<25} {'By partition':<25}")

🪣 Bucketing

🎯 Dividing Data into Buckets

Bucketing divides data into a fixed number of files (buckets) based on a hash of a column. It enables efficient sampling, joins, and aggregation.

Bucketing

Hash: Column → bucket. Fixed: Number of buckets. Sampling: Efficient. Joins: Bucket-to-bucket. Sort: Within bucket. Files: One per bucket.

🪣 Hash🪣 Fixed Buckets🪣 Sampling🪣 Joins
# Bucketing in Hive print("=== What is Bucketing? ===") print(""" Bucketing: - Divides data into fixed number of buckets - Based on hash of a column - Each bucket is a file - Enables efficient operations Hash function: bucket = hash(column) % num_buckets Example: CLUSTERED BY (user_id) INTO 4 BUCKETS hash(101) % 4 = 1 → Bucket 1 hash(102) % 4 = 2 → Bucket 2 hash(103) % 4 = 3 → Bucket 3 hash(104) % 4 = 0 → Bucket 0 """) print("\n=== Creating Bucketed Table ===") print(""" CREATE TABLE users ( id INT, name STRING, age INT, city STRING ) CLUSTERED BY (id) INTO 4 BUCKETS STORED AS PARQUET; -- With sorting CREATE TABLE users ( id INT, name STRING, age INT, city STRING ) CLUSTERED BY (id) SORTED BY (age ASC) INTO 4 BUCKETS STORED AS PARQUET; Configuration: hive.enforce.bucketing = true (Hive 1.x) -- Hive 2.x+ auto-enforces """) print("\n=== Bucketing vs Partitioning ===") print(f"{'Aspect':<25} {'Partitioning':<25} {'Bucketing':<25}") print("-" * 75) print(f"{'Based on':<25} {'Column value':<25} {'Hash of column':<25}") print(f"{'Number':<25} {'Dynamic':<25} {'Fixed':<25}") print(f"{'Directory':<25} {'Subdirectory':<25} {'File':<25}") print(f"{'Cardinality':<25} {'Low':<25} {'High':<25}") print(f"{'Use case':<25} {'Pruning':<25} {'Sampling, joins':<25}") print(f"{'Combined':<25} {'Yes (partition + bucket)':<25} {'Yes':<25}") print("\n=== Benefits of Bucketing ===") print(""" 1. EFFICIENT SAMPLING - Sample 1 bucket instead of full table - TABLESAMPLE(BUCKET 1 OUT OF 4) 2. EFFICIENT JOINS - Bucket-to-bucket join - Same bucketing columns - Map-side join 3. EFFICIENT AGGREGATION - Partial aggregation per bucket - Combine results 4. SORTED BUCKETS - Sorted within bucket - Efficient merge 5. REDUCED SHUFFLE - Data already partitioned - Less network traffic """) # IMPLEMENTATION print("\n=== Simulating Bucketing ===") class BucketedTable: """Simulate Hive bucketed table.""" def __init__(self, name, columns, bucket_col, num_buckets): self.name = name self.columns = columns self.bucket_col = bucket_col self.num_buckets = num_buckets self.buckets = [[] for _ in range(num_buckets)] def insert(self, rows): """Insert rows into buckets.""" for row in rows: bucket_id = hash(row[self.bucket_col]) % self.num_buckets self.buckets[bucket_id].append(row) def sample(self, bucket_ids): """Sample specific buckets.""" result = [] for bid in bucket_ids: if 0 <= bid < self.num_buckets: result.extend(self.buckets[bid]) return result def bucket_join(self, other): """Bucket-to-bucket join.""" result = [] for i in range(self.num_buckets): for left_row in self.buckets[i]: for right_row in other.buckets[i]: if left_row[self.bucket_col] == right_row[other.bucket_col]: result.append({**left_row, **right_row}) return result def get_bucket_info(self): return [len(b) for b in self.buckets] # Demo print("=== Bucketing Demo ===") # Create bucketed table users = BucketedTable('users', ['id', 'name', 'age'], 'id', num_buckets=4) # Insert data data = [ {'id': 1, 'name': 'Alice', 'age': 30}, {'id': 2, 'name': 'Bob', 'age': 25}, {'id': 3, 'name': 'Charlie', 'age': 35}, {'id': 4, 'name': 'Diana', 'age': 28}, {'id': 5, 'name': 'Eve', 'age': 32}, {'id': 6, 'name': 'Frank', 'age': 40}, {'id': 7, 'name': 'Grace', 'age': 27}, {'id': 8, 'name': 'Henry', 'age': 33}, ] users.insert(data) print(f"Bucket distribution: {users.get_bucket_info()}") # Show buckets print("\nBucket contents:") for i, bucket in enumerate(users.buckets): print(f" Bucket {i}: {[r['name'] for r in bucket]}") # Sampling print("\n=== Sampling (Bucket 1) ===") sample = users.sample([1]) for row in sample: print(f" {row}") # Bucket join print("\n=== Bucket Join ===") orders = BucketedTable('orders', ['user_id', 'order_id', 'amount'], 'user_id', num_buckets=4) orders.insert([ {'user_id': 1, 'order_id': 101, 'amount': 100}, {'user_id': 2, 'order_id': 102, 'amount': 200}, {'user_id': 3, 'order_id': 103, 'amount': 150}, {'user_id': 1, 'order_id': 104, 'amount': 300}, ]) # Need to re-bucket by user_id for join # For demo, let's just show the concept print("Note: For bucket join, both tables must be bucketed by the join key.") print("Users bucketed by id, Orders bucketed by user_id.") print("If bucketing matches, join is map-side (no shuffle).") # BUCKETING BEST PRACTICES print("\n=== Bucketing Best Practices ===") print(""" 1. CHOOSE BUCKET COLUMN WISELY - High cardinality - Used in joins - Evenly distributed 2. NUMBER OF BUCKETS - Power of 2 (efficient hashing) - Based on data size - 256-1024 for large tables 3. USE WITH PARTITIONING - Partition by date - Bucket by user_id - Best of both 4. ENABLE BUCKETING - hive.enforce.bucketing=true (Hive 1.x) - Auto in Hive 2.x+ 5. SORTED BUCKETS - CLUSTERED BY ... SORTED BY ... - Efficient merge joins 6. MATCH BUCKETING FOR JOINS - Same column - Same number of buckets - Enables map-side join """) # BUCKETING SUMMARY print("\n=== Bucketing Summary ===") print(f"{'Feature':<25} {'Description':<40}") print("-" * 65) print(f"{'Purpose':<25} {'Divide into fixed buckets':<40}") print(f"{'Based on':<25} {'Hash of column':<40}") print(f"{'Number':<25} {'Fixed':<40}") print(f"{'Storage':<25} {'One file per bucket':<40}") print(f"{'Sampling':<25} {'Efficient':<40}") print(f"{'Joins':<25} {'Bucket-to-bucket':<40}") print(f"{'Combined with':<25} {'Partitioning':<40}")

📁 File Formats

🎯 Storing Data in Hive

Hive supports many file formats: text, sequence, Avro, Parquet, ORC. Each has trade-offs for storage, performance, and compatibility.

File Formats

Text: Human-readable. Sequence: Binary key-value. Avro: Schema evolution. Parquet: Columnar. ORC: Optimized columnar. Recommend: ORC or Parquet.

📁 Text📁 Parquet📁 ORC📁 Avro
# Hive File Formats print("=== File Formats ===") print(""" 1. TEXTFILE - Plain text - Human-readable - No compression (by default) - Slow - Default format 2. SEQUENCEFILE - Binary key-value - Compressed - Hadoop native - Row-based 3. AVRO - Row-based - Schema evolution - Good for write-heavy - Compact 4. PARQUET - Columnar - Compressed - Fast reads - Good for analytics 5. ORC (Optimized Row Columnar) - Columnar - Compressed - Fast reads - Best for Hive - Indexes, statistics 6. JSONFILE - JSON format - Flexible schema - Human-readable - Slower """) print("\n=== Format Comparison ===") print(f"{'Format':<15} {'Storage':<15} {'Speed':<15} {'Schema':<15} {'Use Case':<20}") print("-" * 85) print(f"{'Text':<15} {'Large':<15} {'Slow':<15} {'No':<15} {'Raw data':<20}") print(f"{'Sequence':<15} {'Medium':<15} {'Medium':<15} {'No':<15} {'Intermediate':<20}") print(f"{'Avro':<15} {'Small':<15} {'Medium':<15} {'Yes':<15} {'Write-heavy':<20}") print(f"{'Parquet':<15} {'Small':<15} {'Fast':<15} {'Yes':<15} {'Analytics':<20}") print(f"{'ORC':<15} {'Smallest':<15} {'Fastest':<15} {'Yes':<15} {'Hive analytics':<20}") print(f"{'JSON':<15} {'Large':<15} {'Slow':<15} {'Yes':<15} {'Semi-structured':<20}") print("\n=== Creating Tables with Formats ===") print(""" -- Text CREATE TABLE users_text (...) STORED AS TEXTFILE; -- Parquet CREATE TABLE users_parquet (...) STORED AS PARQUET; -- ORC CREATE TABLE users_orc (...) STORED AS ORC TBLPROPERTIES ("orc.compress"="SNAPPY"); -- Avro CREATE TABLE users_avro (...) STORED AS AVRO TBLPROPERTIES ( 'avro.schema.literal'='{"type":"record", ...}' ); -- Sequence CREATE TABLE users_seq (...) STORED AS SEQUENCEFILE; """) print("\n=== ORC vs Parquet ===") print(f"{'Feature':<25} {'ORC':<25} {'Parquet':<25}") print("-" * 75) print(f"{'Origin':<25} {'Hive':<25} {'Twitter':<25}") print(f"{'Type':<25} {'Columnar':<25} {'Columnar':<25}") print(f"{'Compression':<25} {'Excellent':<25} {'Excellent':<25}") print(f"{'Indexes':<25} {'Built-in':<25} {'Limited':<25}") print(f"{'Statistics':<25} {'Yes':<25} {'Yes':<25}") print(f"{'Predicate pushdown':<25} {'Yes':<25} {'Yes':<25}") print(f"{'Hive support':<25} {'Native':<25} {'Good':<25}") print(f"{'Spark support':<25} {'Good':<25} {'Native':<25}") print(f"{'Ecosystem':<25} {'Hive-first':<25} {'Broader':<25}") # IMPLEMENTATION print("\n=== Simulating File Formats ===") import json import csv from io import StringIO class FileFormatConverter: """Simulate different file formats.""" def __init__(self, data): self.data = data def to_text(self): """Text format (CSV-like).""" output = StringIO() if self.data: writer = csv.DictWriter(output, fieldnames=self.data[0].keys()) writer.writeheader() writer.writerows(self.data) return output.getvalue() def to_json(self): """JSON format.""" return json.dumps(self.data, indent=2) def to_parquet_like(self): """Simulate Parquet (columnar).""" # Columnar representation columns = {} if self.data: for key in self.data[0].keys(): columns[key] = [row[key] for row in self.data] return columns def to_orc_like(self): """Simulate ORC (columnar + stats).""" columns = self.to_parquet_like() stats = {} for col, values in columns.items(): if values and isinstance(values[0], (int, float)): stats[col] = { 'min': min(values), 'max': max(values), 'count': len(values), 'sum': sum(values), } return {'columns': columns, 'stats': stats} def size_comparison(self): """Compare sizes.""" text_size = len(self.to_text()) json_size = len(self.to_json()) parquet_size = text_size * 0.3 # Estimate orc_size = text_size * 0.25 # Estimate return { 'Text': text_size, 'JSON': json_size, 'Parquet': parquet_size, 'ORC': orc_size, } # Demo print("=== File Format Demo ===") data = [ {'id': 1, 'name': 'Alice', 'age': 30, 'salary': 50000}, {'id': 2, 'name': 'Bob', 'age': 25, 'salary': 60000}, {'id': 3, 'name': 'Charlie', 'age': 35, 'salary': 75000}, ] converter = FileFormatConverter(data) print("Text format:") print(converter.to_text()) print("JSON format:") print(converter.to_json()) print("Columnar (Parquet-like):") col_data = converter.to_parquet_like() for col, values in col_data.items(): print(f" {col}: {values}") print("ORC-like with stats:") orc_data = converter.to_orc_like() print(f" Columns: {list(orc_data['columns'].keys())}") print(f" Stats: {orc_data['stats']}") print("\nSize comparison:") sizes = converter.size_comparison() for fmt, size in sizes.items(): print(f" {fmt}: {size:.0f} bytes") # FORMAT BEST PRACTICES print("\n=== File Format Best Practices ===") print(""" 1. USE ORC OR PARQUET - Best compression - Fast reads - Columnar 2. AVOID TEXT FOR PRODUCTION - Large - Slow - No schema 3. USE AVRO FOR STREAMING - Schema evolution - Row-based - Good for writes 4. COMPRESS - Snappy: Fast - Zlib: Better - Zstd: Best 5. PARTITION + COLUMNAR - Best performance - Partition pruning - Column pruning 6. MATCH FORMAT TO USE CASE - Analytics: ORC/Parquet - Streaming: Avro - Raw: Text - Intermediate: Sequence """) # FORMAT SUMMARY print("\n=== File Format Summary ===") print(f"{'Format':<15} {'Type':<15} {'Compression':<15} {'Best For':<25}") print("-" * 70) print(f"{'Text':<15} {'Row':<15} {'None':<15} {'Raw data':<25}") print(f"{'Sequence':<15} {'Row':<15} {'Yes':<15} {'Intermediate':<25}") print(f"{'Avro':<15} {'Row':<15} {'Yes':<15} {'Write-heavy':<25}") print(f"{'Parquet':<15} {'Columnar':<15} {'Yes':<15} {'Analytics':<25}") print(f"{'ORC':<15} {'Columnar':<15} {'Yes':<15} {'Hive analytics':<25}") print(f"{'JSON':<15} {'Row':<15} {'No':<15} {'Semi-structured':<25}")

🔤 SerDe

🎯 Serialization and Deserialization

SerDe (Serializer/Deserializer) tells Hive how to read and write data. It converts between Hive's internal row format and the file format.

SerDe

Serialize: Row → bytes. Deserialize: Bytes → row. Built-in: Text, JSON, Avro, ORC. Custom: User-defined. Format: Row format. Properties: Delimiters, escaping.

🔤 Serialize🔤 Deserialize🔤 Row Format🔤 Custom
# SerDe — Serialization/Deserialization print("=== What is SerDe? ===") print(""" SerDe: - Serializer/Deserializer - Converts between: - Hive row format (in memory) - File format (on disk) Serialize: - Row → bytes - When writing Deserialize: - Bytes → row - When reading Built-in SerDes: - LazySimpleSerDe (text) - JsonSerDe - AvroSerDe - OrcSerDe - ParquetSerDe - RegexSerDe """) print("\n=== Row Format ===") print(""" Row Format specifies: - How rows are delimited - How columns are delimited - How complex types are represented Example: ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' COLLECTION ITEMS TERMINATED BY '|' MAP KEYS TERMINATED BY ':' LINES TERMINATED BY '\\n' NULL DEFINED AS 'NULL' This uses LazySimpleSerDe. """) print("\n=== Creating Tables with SerDe ===") print(""" -- Text with delimiters (LazySimpleSerDe) CREATE TABLE users ( id INT, name STRING, age INT ) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' STORED AS TEXTFILE; -- JSON (JsonSerDe) CREATE TABLE users_json ( id INT, name STRING, age INT ) ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' STORED AS TEXTFILE; -- Regex (RegexSerDe) CREATE TABLE logs ( ip STRING, timestamp STRING, request STRING, status INT ) ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe' WITH SERDEPROPERTIES ( 'input.regex'='^(\\\\S+) (\\\\S+) (\\\\S+) (\\\\d+)' ) STORED AS TEXTFILE; -- Avro CREATE TABLE users_avro ( id INT, name STRING, age INT ) ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe' STORED AS AVRO; -- Custom SerDe CREATE TABLE custom ( ... ) ROW FORMAT SERDE 'com.example.MySerDe'; """) # IMPLEMENTATION print("\n=== Simulating SerDe ===") class SimpleSerDe: """Simulate LazySimpleSerDe.""" def __init__(self, field_delim=',', line_delim='\n'): self.field_delim = field_delim self.line_delim = line_delim def deserialize(self, text): """Text → rows.""" rows = [] lines = text.strip().split(self.line_delim) for line in lines: fields = line.split(self.field_delim) rows.append(fields) return rows def serialize(self, rows): """Rows → text.""" lines = [] for row in rows: lines.append(self.field_delim.join(str(f) for f in row)) return self.line_delim.join(lines) class JSONSerDe: """Simulate JsonSerDe.""" def deserialize(self, text): """JSON → rows.""" data = json.loads(text) if isinstance(data, list): return data return [data] def serialize(self, rows): """Rows → JSON.""" return json.dumps(rows, indent=2) class RegexSerDe: """Simulate RegexSerDe.""" def __init__(self, pattern): import re self.pattern = re.compile(pattern) def deserialize(self, text): """Text → rows using regex.""" rows = [] for line in text.strip().split('\n'): match = self.pattern.match(line) if match: rows.append(list(match.groups())) return rows # Demo print("=== SerDe Demo ===") # CSV SerDe print("\n1. LazySimpleSerDe (CSV):") csv_serde = SimpleSerDe(field_delim=',') csv_text = """1,Alice,30 2,Bob,25 3,Charlie,35""" rows = csv_serde.deserialize(csv_text) print(f" Deserialized: {rows}") serialized = csv_serde.serialize(rows) print(f" Serialized: {serialized}") # JSON SerDe print("\n2. JsonSerDe:") json_serde = JSONSerDe() json_text = '[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]' rows = json_serde.deserialize(json_text) print(f" Deserialized: {rows}") # Regex SerDe print("\n3. RegexSerDe:") regex_serde = RegexSerDe(r'^(\S+) (\S+) (\S+) (\d+)$') log_text = """192.168.1.1 - - 200 10.0.0.1 - - 404""" rows = regex_serde.deserialize(log_text) print(f" Deserialized: {rows}") # SERDE PROPERTIES print("\n=== SerDe Properties ===") print(""" Common SerDe properties: LazySimpleSerDe: serialization.format=, field.delim=, line.delim=\\n collection.delim=| mapkey.delim=: escape.delim=\\\\ null.format=NULL JsonSerDe: (none — JSON is self-describing) RegexSerDe: input.regex=... output.format.string=... AvroSerDe: avro.schema.literal=... avro.schema.url=... """) # SERDE SUMMARY print("\n=== SerDe Summary ===") print(f"{'SerDe':<25} {'Format':<25} {'Use Case':<25}") print("-" * 75) print(f"{'LazySimpleSerDe':<25} {'Delimited text':<25} {'CSV, TSV':<25}") print(f"{'JsonSerDe':<25} {'JSON':<25} {'Semi-structured':<25}") print(f"{'RegexSerDe':<25} {'Regex':<25} {'Logs':<25}") print(f"{'AvroSerDe':<25} {'Avro':<25} {'Schema evolution':<25}") print(f"{'OrcSerDe':<25} {'ORC':<25} {'Analytics':<25}") print(f"{'ParquetSerDe':<25} {'Parquet':<25} {'Analytics':<25}")

⚡ Optimization

🎯 Making Hive Fast

Hive optimization focuses on reducing data scanned, minimizing shuffle, and using efficient execution engines. Partitioning, bucketing, and ORC are key techniques.

Optimization Techniques

Partitioning: Prune data. Bucketing: Efficient joins. ORC: Columnar format. Tez: DAG engine. Vectorization: Batch processing. CBO: Cost-based optimizer.

⚡ Partitioning⚡ Bucketing⚡ ORC⚡ Tez
# Hive Optimization print("=== 1. Partitioning ===") print(""" Partition pruning: - Filter on partition columns - Hive skips irrelevant partitions Example: SELECT * FROM events WHERE year=2024 AND month=1; Hive only reads year=2024/month=1 partition. Best practices: - Partition by commonly filtered columns - Low cardinality - Time-based (year, month, day) - Avoid over-partitioning """) print("\n=== 2. Bucketing ===") print(""" Bucketing: - Fixed number of buckets - Hash of column - Efficient joins - Sampling Example: CLUSTERED BY (user_id) INTO 256 BUCKETS Benefits: - Bucket-to-bucket join - Map-side join - Efficient sampling """) print("\n=== 3. File Formats ===") print(""" Use ORC or Parquet: - Columnar - Compressed - Predicate pushdown - Column pruning ORC features: - Indexes - Statistics - Bloom filters Configuration: CREATE TABLE ... STORED AS ORC TBLPROPERTIES ("orc.compress"="SNAPPY"); """) print("\n=== 4. Execution Engine ===") print(""" MapReduce (original): - Disk-based - Slow - Legacy Tez (recommended): - DAG execution - In-memory - 10x faster - Set: hive.execution.engine=tez Spark: - In-memory - Unified - Set: hive.execution.engine=spark """) print("\n=== 5. Vectorization ===") print(""" Vectorized query execution: - Process 1024 rows at a time - Better CPU utilization - Faster Enable: SET hive.vectorized.execution.enabled=true; SET hive.vectorized.execution.reduce.enabled=true; Requirements: - ORC format - Supported data types """) print("\n=== 6. Cost-Based Optimizer (CBO) ===") print(""" CBO: - Uses statistics - Chooses join order - Chooses join type - Better plans Enable: SET hive.cbo.enable=true; SET hive.compute.query.using.stats=true; SET hive.stats.fetch.column.stats=true; Statistics: ANALYZE TABLE users COMPUTE STATISTICS; ANALYZE TABLE users COMPUTE STATISTICS FOR COLUMNS; """) print("\n=== 7. Join Optimization ===") print(""" 1. MAP-SIDE JOIN - Small table fits in memory - Broadcast to mappers SET hive.auto.convert.join=true; SET hive.mapjoin.smalltable.filesize=25000000; 2. BUCKET MAP JOIN - Both tables bucketed - Same bucket column SET hive.optimize.bucketmapjoin=true; 3. SORT-MERGE-BUCKET JOIN - Both tables sorted and bucketed SET hive.optimize.bucketmapjoin.sortedmerge=true; 4. SKEWED JOIN - Handle skewed keys SET hive.optimize.skewjoin=true; """) print("\n=== 8. Other Optimizations ===") print(""" - Predicate pushdown - Partition pruning - Column pruning - Parallel execution - JVM reuse - Compression - Avoid small files - Use LIMIT - Use EXPLAIN """) # IMPLEMENTATION print("\n=== Optimization Comparison ===") import time class OptimizedQuery: """Compare optimized vs non-optimized queries.""" def __init__(self, data_size_gb): self.data_size_gb = data_size_gb def without_optimization(self): """Full table scan.""" print("\n=== Without Optimization ===") # Scan all data scan_time = self.data_size_gb * 10 # 10 sec per GB shuffle_time = self.data_size_gb * 5 total = scan_time + shuffle_time print(f" Data size: {self.data_size_gb} GB") print(f" Scan time: {scan_time:.1f} s") print(f" Shuffle time: {shuffle_time:.1f} s") print(f" Total: {total:.1f} s") return total def with_partitioning(self, partition_pruning=0.1): """With partition pruning.""" print("\n=== With Partitioning ===") effective_size = self.data_size_gb * partition_pruning scan_time = effective_size * 10 shuffle_time = effective_size * 5 total = scan_time + shuffle_time print(f" Data size: {self.data_size_gb} GB") print(f" After pruning: {effective_size:.1f} GB") print(f" Scan time: {scan_time:.1f} s") print(f" Shuffle time: {shuffle_time:.1f} s") print(f" Total: {total:.1f} s") return total def with_orc(self, compression=0.3): """With ORC compression.""" print("\n=== With ORC ===") effective_size = self.data_size_gb * compression scan_time = effective_size * 5 # ORC is faster shuffle_time = effective_size * 3 total = scan_time + shuffle_time print(f" Data size: {self.data_size_gb} GB") print(f" After ORC: {effective_size:.1f} GB") print(f" Scan time: {scan_time:.1f} s") print(f" Shuffle time: {shuffle_time:.1f} s") print(f" Total: {total:.1f} s") return total def with_tez(self, speedup=5): """With Tez engine.""" print("\n=== With Tez ===") base_time = self.without_optimization() tez_time = base_time / speedup print(f" Base time: {base_time:.1f} s") print(f" Tez time: {tez_time:.1f} s") print(f" Speedup: {speedup}x") return tez_time def with_all(self): """With all optimizations.""" print("\n=== With All Optimizations ===") # Partitioning effective_size = self.data_size_gb * 0.1 # ORC effective_size *= 0.3 # Tez scan_time = effective_size * 2 # Fast shuffle_time = effective_size * 1 total = (scan_time + shuffle_time) / 5 print(f" Data size: {self.data_size_gb} GB") print(f" Effective size: {effective_size:.2f} GB") print(f" Total: {total:.2f} s") return total # Compare print("\n=== Optimization Comparison ===") query = OptimizedQuery(data_size_gb=1000) t1 = query.without_optimization() t2 = query.with_partitioning() t3 = query.with_orc() t4 = query.with_all() print(f"\n=== Speedup Summary ===") print(f" Baseline: {t1:.0f} s") print(f" Partitioning: {t2:.0f} s ({t1/t2:.1f}x)") print(f" ORC: {t3:.0f} s ({t1/t3:.1f}x)") print(f" All: {t4:.0f} s ({t1/t4:.1f}x)") # OPTIMIZATION CHECKLIST print("\n=== Optimization Checklist ===") print(""" ✓ Partition by commonly filtered columns ✓ Bucket by join columns ✓ Use ORC or Parquet ✓ Enable Tez or Spark ✓ Enable vectorization ✓ Enable CBO ✓ Use map-side joins ✓ Handle skew ✓ Compress data ✓ Avoid small files ✓ Use EXPLAIN ✓ Analyze statistics """) # OPTIMIZATION SUMMARY print("\n=== Optimization Summary ===") print(f"{'Technique':<25} {'Impact':<30} {'Effort':<15}") print("-" * 70) print(f"{'Partitioning':<25} {'High (10-100x)':<30} {'Low':<15}") print(f"{'Bucketing':<25} {'Medium (2-10x)':<30} {'Medium':<15}") print(f"{'ORC/Parquet':<25} {'High (3-10x)':<30} {'Low':<15}") print(f"{'Tez/Spark':<25} {'High (5-10x)':<30} {'Low':<15}") print(f"{'Vectorization':<25} {'Medium (2-5x)':<30} {'Low':<15}") print(f"{'CBO':<25} {'Medium (2-5x)':<30} {'Low':<15}") print(f"{'Map-side join':<25} {'High (10-100x)':<30} {'Medium':<15}")

🔧 UDFs & Transform

🎯 Custom Functions

UDFs (User-Defined Functions) extend HiveQL with custom logic. Hive supports UDFs (row), UDAFs (aggregate), and UDTFs (table-generating).

UDF Types

UDF: Row → row. UDAF: Aggregate. UDTF: Table-generating. Languages: Java, Python. Transform: Streaming. Register: CREATE FUNCTION.

🔧 UDF🔧 UDAF🔧 UDTF🔧 Transform
# UDFs in Hive print("=== Types of UDFs ===") print(""" 1. UDF (User-Defined Function) - Row → row - One input, one output - Example: UPPER, LOWER, custom SELECT my_udf(name) FROM users; 2. UDAF (User-Defined Aggregate Function) - Many rows → one value - Aggregation - Example: SUM, AVG, custom SELECT my_udaf(salary) FROM users; 3. UDTF (User-Defined Table-Generating Function) - One row → many rows - Explode-like - Example: EXPLODE, custom SELECT my_udtf(skills) FROM users; """) print("\n=== Built-in UDFs ===") print(""" String: UPPER, LOWER, TRIM, LENGTH, CONCAT, SUBSTR, REGEXP_REPLACE Math: ABS, ROUND, FLOOR, CEIL, SQRT, POW Date: YEAR, MONTH, DAY, DATE_ADD, DATEDIFF, CURRENT_DATE Collection: SIZE, ARRAY_CONTAINS, SORT_ARRAY, MAP_KEYS Conditional: IF, CASE, COALESCE, NVL Type conversion: CAST, BINARY JSON: GET_JSON_OBJECT """) print("\n=== Creating Custom UDF (Java) ===") print(""" package com.example.hive.udf; import org.apache.hadoop.hive.ql.exec.UDF; import org.apache.hadoop.io.Text; public class MyUpper extends UDF { public Text evaluate(Text input) { if (input == null) return null; return new Text(input.toString().toUpperCase()); } } -- Register ADD JAR /path/to/udf.jar; CREATE TEMPORARY FUNCTION my_upper AS 'com.example.hive.udf.MyUpper'; -- UseSELECT my_upper(name) FROM users; """) print("\n=== UDF in Python (Transform) ===") print(""" -- Using TRANSFORM with Python SELECT TRANSFORM(name, age) USING 'python3 /path/to/script.py' AS (new_name, new_age) FROM users; -- script.py import sys for line in sys.stdin: line = line.strip() name, age = line.split('\\t') new_name = name.upper() new_age = int(age) + 1 print(f'{new_name}\\t{new_age}') """) # IMPLEMENTATION print("\n=== Simulating UDFs ===") class HiveUDF: """Base class for UDFs.""" def evaluate(self, *args): raise NotImplementedError class UpperUDF(HiveUDF): """Uppercase UDF.""" def evaluate(self, value): return value.upper() if value else None class ConcatUDF(HiveUDF): """Concatenate UDF.""" def evaluate(self, *args): return ''.join(str(a) for a in args if a) class AgeBucketUDF(HiveUDF): """Age bucket UDF.""" def evaluate(self, age): if age < 18: return 'minor' elif age < 65: return 'adult' else: return 'senior' class SumUDAF: """Sum aggregate function.""" def __init__(self): self.total = 0 def add(self, value): self.total += value def result(self): return self.total class ExplodeUDTF: """Explode UDTF.""" def evaluate(self, array): for item in array: yield item # Demo print("=== UDF Demo ===") # Register UDFs udfs = { 'my_upper': UpperUDF(), 'my_concat': ConcatUDF(), 'age_bucket': AgeBucketUDF(), } # Sample data users = [ {'id': 1, 'name': 'alice', 'age': 30}, {'id': 2, 'name': 'bob', 'age': 15}, {'id': 3, 'name': 'charlie', 'age': 70}, ] print("1. my_upper(name):") for user in users: result = udfs['my_upper'].evaluate(user['name']) print(f" {user['name']} → {result}") print("\n2. age_bucket(age):") for user in users: bucket = udfs['age_bucket'].evaluate(user['age']) print(f" {user['name']} ({user['age']}) → {bucket}") print("\n3. my_concat(name, age):") for user in users: result = udfs['my_concat'].evaluate(user['name'], '-', user['age']) print(f" → {result}") # UDAF print("\n4. UDAF (sum):") sum_udaf = SumUDAF() for user in users: sum_udaf.add(user['age']) print(f" Total age: {sum_udaf.result()}") # UDTF print("\n5. UDTF (explode):") explode = ExplodeUDTF() skills = ['python', 'sql', 'spark'] for skill in explode.evaluate(skills): print(f" {skill}") # UDF BEST PRACTICES print("\n=== UDF Best Practices ===") print(""" 1. PREFER BUILT-IN FUNCTIONS - Faster - Tested - Optimized 2. USE SIMPLE TYPES - Primitive types - Avoid complex objects 3. HANDLE NULLS - Check for null - Return null appropriately 4. OPTIMIZE PERFORMANCE - Avoid heavy computation - Cache results - Use vectorization 5. TEST THOROUGHLY - Unit tests - Edge cases - Null handling 6. DOCUMENT - Purpose - Parameters - Return value 7. PACKAGE PROPERLY - JAR file - Dependencies - Version control 8. USE TRANSFORM FOR PYTHON - Simpler than Java UDF - Good for prototyping - Slower than Java """) # UDF SUMMARY print("\n=== UDF Summary ===") print(f"{'Type':<25} {'Input':<20} {'Output':<20} {'Example':<20}") print("-" * 85) print(f"{'UDF':<25} {'One row':<20} {'One value':<20} {'UPPER':<20}") print(f"{'UDAF':<25} {'Many rows':<20} {'One value':<20} {'SUM':<20}") print(f"{'UDTF':<25} {'One row':<20} {'Many rows':<20} {'EXPLODE':<20}") print(f"{'Transform':<25} {'Stream':<20} {'Stream':<20} {'Python':<20}")

🔐 ACID Transactions

🎯 Transactions in Hive

ACID (Atomicity, Consistency, Isolation, Durability) transactions allow Hive to support updates, deletes, and concurrent reads/writes. Introduced in Hive 0.14.

ACID in Hive

Atomicity: All or nothing. Consistency: Valid state. Isolation: Concurrent safety. Durability: Persisted. Operations: INSERT, UPDATE, DELETE. Format: ORC only.

🔐 Atomicity🔐 Consistency🔐 Isolation🔐 Durability
# ACID Transactions in Hive print("=== What is ACID? ===") print(""" ACID: - Atomicity: All or nothing - Consistency: Valid state transitions - Isolation: Concurrent transactions don't interfere - Durability: Committed changes persist Hive ACID: - Introduced in Hive 0.14 - Supports UPDATE, DELETE - Requires ORC format - Requires bucketing - Requires transactions enabled """) print("\n=== Enabling ACID ===") print(""" -- Configuration SET hive.support.concurrency=true; SET hive.txn.manager=org.apache.hadoop.hive.ql.lockmgr.DbTxnManager; SET hive.compactor.initiator.on=true; SET hive.compactor.worker.threads=1; -- Create ACID table CREATE TABLE users ( id INT, name STRING, age INT, city STRING ) CLUSTERED BY (id) INTO 4 BUCKETS STORED AS ORC TBLPROPERTIES ('transactional'='true'); -- Insert INSERT INTO users VALUES (1, 'Alice', 30, 'NYC'); -- Update UPDATE users SET age = 31 WHERE id = 1; -- Delete DELETE FROM users WHERE id = 1; -- Merge (upsert) MERGE INTO users AS target USING updates AS source ON target.id = source.id WHEN MATCHED THEN UPDATE SET age = source.age WHEN NOT MATCHED THEN INSERT VALUES (source.id, source.name, source.age, source.city); """) print("\n=== ACID Requirements ===") print(""" 1. TABLE FORMAT - ORC only - Not text, Parquet, Avro 2. BUCKETING - Table must be bucketed - CLUSTERED BY 3. TABLE PROPERTY - 'transactional'='true' 4. CONFIGURATION - Concurrency enabled - Transaction manager - Compactor 5. HIVE VERSION - Hive 0.14+ - Hive 2.x+ recommended """) print("\n=== ACID Operations ===") print(""" INSERT: INSERT INTO users VALUES (...); INSERT INTO users SELECT ...; UPDATE: UPDATE users SET age = 31 WHERE id = 1; DELETE: DELETE FROM users WHERE id = 1; MERGE: MERGE INTO users AS t USING updates AS s ON t.id = s.id WHEN MATCHED THEN UPDATE SET ... WHEN NOT MATCHED THEN INSERT VALUES ...; """) # IMPLEMENTATION print("\n=== Simulating ACID ===") import threading import time class ACIDTable: """Simulate Hive ACID table.""" def __init__(self, name): self.name = name self.data = {} # id -> row self.locks = {} self.transaction_log = [] def begin_transaction(self, txn_id): """Begin transaction.""" self.transaction_log.append({ 'txn_id': txn_id, 'action': 'BEGIN', 'timestamp': time.time(), }) print(f" Txn {txn_id}: BEGIN") def insert(self, txn_id, row): """Insert row.""" row_id = row['id'] if row_id in self.data: print(f" Txn {txn_id}: INSERT failed (duplicate id)") return False self.data[row_id] = row self.transaction_log.append({ 'txn_id': txn_id, 'action': 'INSERT', 'row': row, 'timestamp': time.time(), }) print(f" Txn {txn_id}: INSERT {row}") return True def update(self, txn_id, row_id, updates): """Update row.""" if row_id not in self.data: print(f" Txn {txn_id}: UPDATE failed (row not found)") return False old_row = self.data[row_id].copy() self.data[row_id].update(updates) self.transaction_log.append({ 'txn_id': txn_id, 'action': 'UPDATE', 'row_id': row_id, 'old': old_row, 'new': self.data[row_id], 'timestamp': time.time(), }) print(f" Txn {txn_id}: UPDATE {row_id} → {updates}") return True def delete(self, txn_id, row_id): """Delete row.""" if row_id not in self.data: print(f" Txn {txn_id}: DELETE failed (row not found)") return False old_row = self.data.pop(row_id) self.transaction_log.append({ 'txn_id': txn_id, 'action': 'DELETE', 'row_id': row_id, 'old': old_row, 'timestamp': time.time(), }) print(f" Txn {txn_id}: DELETE {row_id}") return True def commit(self, txn_id): """Commit transaction.""" self.transaction_log.append({ 'txn_id': txn_id, 'action': 'COMMIT', 'timestamp': time.time(), }) print(f" Txn {txn_id}: COMMIT") def rollback(self, txn_id): """Rollback transaction.""" # Undo operations in reverse order for entry in reversed(self.transaction_log): if entry['txn_id'] != txn_id: continue if entry['action'] == 'INSERT': row_id = entry['row']['id'] if row_id in self.data: del self.data[row_id] elif entry['action'] == 'UPDATE': self.data[entry['row_id']] = entry['old'] elif entry['action'] == 'DELETE': self.data[entry['row_id']] = entry['old'] self.transaction_log.append({ 'txn_id': txn_id, 'action': 'ROLLBACK', 'timestamp': time.time(), }) print(f" Txn {txn_id}: ROLLBACK") def select(self): """Select all rows.""" return list(self.data.values()) # Demo print("=== ACID Demo ===") table = ACIDTable('users') # Transaction 1: Insert print("\n--- Transaction 1 ---") table.begin_transaction(1) table.insert(1, {'id': 1, 'name': 'Alice', 'age': 30}) table.insert(2, {'id': 2, 'name': 'Bob', 'age': 25}) table.commit(1) print(f"Data after T1: {table.select()}") # Transaction 2: Update + rollback print("\n--- Transaction 2 (rollback) ---") table.begin_transaction(2) table.update(2, 1, {'age': 31}) table.update(2, 2, {'age': 26}) print(f"Data before rollback: {table.select()}") table.rollback(2) print(f"Data after rollback: {table.select()}") # Transaction 3: Update + commit print("\n--- Transaction 3 (commit) ---") table.begin_transaction(3) table.update(3, 1, {'age': 31}) table.commit(3) print(f"Data after T3: {table.select()}") # Transaction 4: Delete print("\n--- Transaction 4 (delete) ---") table.begin_transaction(4) table.delete(4, 2) table.commit(4) print(f"Data after T4: {table.select()}") # ACID SUMMARY print("\n=== ACID Summary ===") print(f"{'Property':<20} {'Description':<40}") print("-" * 65) print(f"{'Atomicity':<20} {'All or nothing':<40}") print(f"{'Consistency':<20} {'Valid state':<40}") print(f"{'Isolation':<20} {'Concurrent safety':<40}") print(f"{'Durability':<20} {'Persisted':<40}") print("\n=== Hive ACID Requirements ===") print(f"{'Requirement':<25} {'Value':<30}") print("-" * 60) print(f"{'Format':<25} {'ORC':<30}") print(f"{'Bucketing':<25} {'Required':<30}") print(f"{'Table property':<25} {'transactional=true':<30}") print(f"{'Concurrency':<25} {'Enabled':<30}") print(f"{'Version':<25} {'0.14+':<30}")

⚖️ Hive vs RDBMS

🎯 Comparing Hive and Traditional Databases

Hive and RDBMS both use SQL, but they're designed for different workloads. Hive is for batch analytics on Big Data; RDBMS is for transactional workloads.

Comparison

Hive: Batch, PB scale, schema-on-read. RDBMS: OLTP, GB scale, schema-on-write. Latency: Minutes vs ms. Updates: Limited vs full. Storage: HDFS vs local. Use: Analytics vs transactions.

⚖️ Batch vs OLTP⚖️ PB vs GB⚖️ Schema-on-read⚖️ Analytics
# Hive vs RDBMS print("=== Hive vs RDBMS ===") print(f"{'Aspect':<25} {'Hive':<25} {'RDBMS':<25}") print("-" * 75) print(f"{'Workload':<25} {'Batch analytics':<25} {'OLTP':<25}") print(f"{'Latency':<25} {'Minutes-hours':<25} {'Milliseconds':<25}") print(f"{'Scale':<25} {'Petabytes':<25} {'Gigabytes':<25}") print(f"{'Storage':<25} {'HDFS':<25} {'Local disk':<25}") print(f"{'Schema':<25} {'Schema-on-read':<25} {'Schema-on-write':<25}") print(f"{'Updates':<25} {'Limited (ACID)':<25} {'Full':<25}") print(f"{'Transactions':<25} {'ACID (limited)':<25} {'Full ACID':<25}") print(f"{'Indexes':<25} {'Limited':<25} {'Extensive':<25}") print(f"{'Constraints':<25} {'Limited':<25} {'Full':<25}") print(f"{'Joins':<25} {'Limited':<25} {'Optimized':<25}") print(f"{'Cost':<25} {'Low':<25} {'High':<25}") print(f"{'Use case':<25} {'Analytics':<25} {'Transactions':<25}") print("\n=== Architecture Comparison ===") print(""" RDBMS: - Single machine (scale up) - Shared memory - ACID transactions - Normalized schema - Indexes for speed Hive: - Distributed cluster (scale out) - Shared nothing - ACID (limited) - Denormalized schema - Partitions for speed """) print("\n=== When to Use Which ===") print(""" USE HIVE WHEN: ✓ Petabytes of data ✓ Batch processing ✓ Analytics/reporting ✓ Unstructured data ✓ Cost-sensitive ✓ Append-only data USE RDBMS WHEN: ✓ Transactions (OLTP) ✓ Low latency ✓ Data integrity ✓ Complex joins ✓ Frequent updates ✓ Small data """) print("\n=== Hive + RDBMS Integration ===") print(""" Common patterns: 1. ETL PIPELINE RDBMS → Sqoop → Hive → Analytics 2. DATA WAREHOUSE RDBMS (source) → Hive (warehouse) 3. REPORTING Hive (aggregate) → RDBMS (serve) 4. REAL-TIME Hive (batch) + RDBMS (serving layer) """) # COMPARISON MATRIX print("\n=== Decision Matrix ===") print(f"{'Requirement':<25} {'Best Choice':<25} {'Alternative':<25}") print("-" * 75) print(f"{'OLTP':<25} {'RDBMS':<25} {'Hive (ACID)':<25}") print(f"{'Analytics':<25} {'Hive':<25} {'RDBMS (warehouse)':<25}") print(f"{'Petabytes':<25} {'Hive':<25} {'—':<25}") print(f"{'Low latency':<25} {'RDBMS':<25} {'HBase':<25}") print(f"{'Updates':<25} {'RDBMS':<25} {'Hive (ACID)':<25}") print(f"{'Cost':<25} {'Hive':<25} {'RDBMS':<25}") print(f"{'Complex joins':<25} {'RDBMS':<25} {'Hive (Tez)':<25}") # SUMMARY print("\n=== Summary ===") print(""" Hive: - SQL on Hadoop - Batch analytics - Petabyte scale - Schema-on-read - Low cost - Limited transactions RDBMS: - Traditional database - OLTP - Gigabyte scale - Schema-on-write - Higher cost - Full ACID They're complementary: - RDBMS for source/serving - Hive for analytics/warehouse """)

💼 Applications

🎯 Hive in the Real World

Hive powers data warehousing, log analysis, and analytics at companies like Facebook, Netflix, and LinkedIn. It's the SQL interface to Big Data.

Applications

Data warehouse: Central analytics. Log analysis: Server logs. BI: Dashboards. ETL: Transformation. Data mining: Patterns. ML: Feature engineering.

💼 Data Warehouse💼 Log Analysis💼 BI💼 ETL
# Hive Applications print("=== 1. Data Warehousing ===") print(""" Central data warehouse on Hadoop: Architecture: Sources → ETL → Hive (warehouse) → BI/Analytics Tables: - Staging (raw) - ODS (operational data store) - DW (data warehouse) - Data marts (subject-specific) ETL: - Sqoop (RDBMS → Hive) - Flume (logs → Hive) - Kafka (stream → Hive) Examples: - Facebook: 300+ PB - Netflix: Recommendations - LinkedIn: Analytics """) print("\n=== 2. Log Analysis ===") print(""" Analyze server/application logs: Use cases: - Error tracking - User behavior - Security analysis - Performance monitoring Tables: CREATE TABLE logs ( timestamp STRING, ip STRING, method STRING, url STRING, status INT, bytes INT ) PARTITIONED BY (year INT, month INT, day INT) STORED AS ORC; Queries: -- Error rate by hour SELECT hour(timestamp), COUNT(*) FROM logs WHERE status >= 500 GROUP BY hour(timestamp); """) print("\n=== 3. Business Intelligence ===") print(""" BI dashboards and reports: Reports: - Daily/weekly/monthly KPIs - Revenue analysis - Customer segmentation - Product performance Tools: - Tableau - Power BI - Looker - Superset Example: -- Daily revenue SELECT date, SUM(amount) as revenue FROM orders GROUP BY date ORDER BY date; """) print("\n=== 4. ETL Pipelines ===") print(""" Extract, Transform, Load: Pipeline: RDBMS → Sqoop → Hive (staging) → Transform → Hive (DW) Transformations: - Cleaning - Joining - Aggregating - Enriching Scheduling: - Oozie - Airflow - Cron """) print("\n=== 5. Data Mining ===") print(""" Pattern discovery: Use cases: - Customer segmentation - Fraud detection - Market basket analysis - Churn prediction Example: -- Customer segmentation SELECT CASE WHEN total_spend > 10000 THEN 'VIP' WHEN total_spend > 1000 THEN 'Regular' ELSE 'Occasional' END as segment, COUNT(*) as customers FROM customer_summary GROUP BY segment; """) print("\n=== 6. Machine Learning Features ===") print(""" Feature engineering: Use cases: - Feature extraction - Feature aggregation - Training data prep - Batch scoring Example: -- User features SELECT user_id, COUNT(*) as order_count, SUM(amount) as total_spend, AVG(amount) as avg_order, DATEDIFF(MAX(date), MIN(date)) as tenure FROM orders GROUP BY user_id; """) # IMPLEMENTATION print("\n=== Practical Example: Log Analyzer ===") class HiveLogAnalyzer: """Simulate Hive log analysis.""" def __init__(self): self.logs = [] def load_logs(self, logs): """Load logs.""" self.logs = logs print(f"Loaded {len(logs)} log entries") def query(self, where=None, group_by=None, aggregates=None): """Execute query.""" data = self.logs # WHERE if where: col, op, val = where if op == '>=': data = [r for r in data if r.get(col, 0) >= val] elif op == '>': data = [r for r in data if r.get(col, 0) > val] elif op == '=': data = [r for r in data if r.get(col) == val] # GROUP BY if group_by: from collections import defaultdict grouped = defaultdict(list) for row in data: grouped[row[group_by]].append(row) result = [] for key, rows in grouped.items(): row_result = {group_by: key} if aggregates: for name, func, col in aggregates: values = [r[col] for r in rows] if func == 'COUNT': row_result[name] = len(values) elif func == 'SUM': row_result[name] = sum(values) elif func == 'AVG': row_result[name] = sum(values) / len(values) result.append(row_result) return sorted(result, key=lambda x: -x.get('count', 0)) return data # Demo print("=== Log Analysis Demo ===") analyzer = HiveLogAnalyzer() # Sample logs logs = [ {'timestamp': '2024-01-01 10:00:00', 'ip': '192.168.1.1', 'method': 'GET', 'url': '/page', 'status': 200, 'bytes': 1234}, {'timestamp': '2024-01-01 10:00:01', 'ip': '192.168.1.2', 'method': 'GET', 'url': '/error', 'status': 404, 'bytes': 0}, {'timestamp': '2024-01-01 10:00:02', 'ip': '192.168.1.1', 'method': 'POST', 'url': '/api', 'status': 500, 'bytes': 0}, {'timestamp': '2024-01-01 10:00:03', 'ip': '192.168.1.3', 'method': 'GET', 'url': '/page', 'status': 200, 'bytes': 2345}, {'timestamp': '2024-01-01 10:00:04', 'ip': '192.168.1.1', 'method': 'GET', 'url': '/page', 'status': 200, 'bytes': 1234}, {'timestamp': '2024-01-01 10:00:05', 'ip': '192.168.1.4', 'method': 'GET', 'url': '/missing', 'status': 404, 'bytes': 0}, ] analyzer.load_logs(logs) print("\n1. Error logs (status >= 400):") errors = analyzer.query(where=('status', '>=', 400)) for log in errors: print(f" {log['timestamp']} {log['status']} {log['url']}") print("\n2. Status code distribution:") dist = analyzer.query(group_by='status', aggregates=[('count', 'COUNT', 'status')]) for row in dist: print(f" Status {row['status']}: {row['count']}") print("\n3. Traffic by IP:") traffic = analyzer.query(group_by='ip', aggregates=[ ('count', 'COUNT', 'ip'), ('bytes', 'SUM', 'bytes') ]) for row in traffic: print(f" {row['ip']}: {row['count']} requests, {row['bytes']} bytes") # APPLICATION SUMMARY print("\n=== Application Summary ===") print(f"{'Domain':<20} {'Use Case':<30} {'Scale':<20}") print("-" * 70) print(f"{'Data Warehouse':<20} {'Central analytics':<30} {'Petabytes':<20}") print(f"{'Log Analysis':<20} {'Error tracking':<30} {'TBs/day':<20}") print(f"{'BI':<20} {'Dashboards':<30} {'Petabytes':<20}") print(f"{'ETL':<20} {'Transformation':<30} {'Petabytes':<20}") print(f"{'Data Mining':<20} {'Patterns':<30} {'Petabytes':<20}") print(f"{'ML Features':<20} {'Feature engineering':<30} {'Petabytes':<20}")

🚀 Future: Hive in Modern Computing

🎯 Where Hive Is Going

Hive is evolving: LLAP, ACID improvements, cloud integration, and competition from Spark SQL, Presto, and cloud warehouses. It remains a key SQL-on-Hadoop engine.

Future Directions

LLAP: Live long and process. Cloud: Managed services. ACID: Better transactions. Competition: Spark, Presto. Lakehouse: Delta/Iceberg. Legacy: Maintenance.

🚀 LLAP☁️ Cloud🔐 ACID🏠 Lakehouse
# Future of Apache Hive print("=== 1. LLAP (Live Long and Process) ===") print(""" LLAP: - Hybrid execution model - Long-lived daemons - In-memory caching - Sub-second queries Benefits: - Interactive queries - Faster than MapReduce - Caching - Columnar Configuration: SET hive.llap.enabled=true; SET hive.llap.execution.mode=all; Use case: - Interactive analytics - BI dashboards - Ad-hoc queries """) print("\n=== 2. Cloud Integration ===") print(""" Hive on cloud: AWS: - EMR (Hive) - Athena (Presto) - Glue (catalog) Google: - Dataproc (Hive) - BigQuery Azure: - HDInsight (Hive) - Synapse Benefits: - Managed - Auto-scaling - Pay per use Future: - Cloud-native - Serverless - Unified catalog """) print("\n=== 3. ACID Improvements ===") print(""" Hive ACID evolution: Hive 1.x: - Basic ACID - ORC only - Bucketing required Hive 2.x: - Better performance - More operations Hive 3.x: - ACID on all tables - Materialized views - Better compaction Future: - Full ACID - Better concurrency - Streaming ingestion """) print("\n=== 4. Competition ===") print(""" Hive faces competition: Spark SQL: - Faster - Unified - DataFrame API Presto/Trino: - Interactive - Federated - Fast Cloud Warehouses: - Snowflake - BigQuery - Redshift Hive response: - LLAP - ACID - Cloud integration - Ecosystem """) print("\n=== 5. Lakehouse Architecture ===") print(""" Hive in lakehouse: Components: - Storage: HDFS or S3 - Table format: Iceberg, Delta, Hudi - Compute: Hive, Spark, Presto - Catalog: Hive Metastore Benefits: - ACID - Schema evolution - Time travel - Unified Future: - Hive as catalog - Multiple engines - Open formats """) print("\n=== 6. New Features ===") print(""" Hive 4.x: - ACID improvements - LLAP enhancements - Better optimizer - Cloud-native Future: - Vectorization - CBO improvements - Materialized views - Streaming """) # TIMELINE print("\n=== Timeline ===") print(""" Past: 2008: Hive started (Facebook) 2010: Apache incubator 2011: Hive 0.7 2013: Hive 0.12 (ORC) 2014: Hive 0.14 (ACID) 2016: Hive 2.0 (LLAP) 2019: Hive 3.0 Present: 2024: Hive 4.0 2025: Maintenance Future: 2030: Hive as catalog? 2035: Lakehouse native? 2040: New paradigms? """) print("\n=== Final Thoughts ===") print(""" Apache Hive is: - SQL on Hadoop - Data warehouse - Petabyte scale - Still relevant The future is: - LLAP for interactive - Cloud-native - ACID everywhere - Lakehouse - Competition from Spark "Hive made Big Data accessible to SQL users." Keep exploring. Keep building. """)