Hive - A Warehousing Solution Over a Map-Reduce Framework

In one breath — the whole paper, compressed

Facebook's data had outgrown what commercial warehouses could economically hold, but Hadoop offered only a raw map-reduce API in which every report was a hand-written Java program. Hive puts a warehouse on top of Hadoop: tables, partitions and hash buckets that are really just HDFS directories and files, a SQL-like language called HiveQL, and a Metastore catalog that holds schemas, storage locations and serialization classes. A driver hands each HiveQL statement to a compiler that parses it, type-checks it against the metastore, builds a tree of logical operators, applies rule-based rewrites such as predicate pushdown and partition pruning, and then cuts the tree into a DAG of map-reduce jobs at repartition and union-all markers. Because arbitrary file formats are read through pluggable SerDes and arbitrary logic can be spliced in as streaming map-reduce scripts, the SQL layer never becomes a cage around the data. At the time of writing, the Facebook instance held several thousand tables and over 700 terabytes, serving more than 100 users and over 5000 queries a day.

Before this paper — the world it landed in

By 2009 the data sets being collected for business intelligence were growing fast enough that traditional warehousing solutions had become prohibitively expensive, and Facebook had already moved its log processing onto Hadoop, an open-source map-reduce implementation running on commodity hardware. The catch was the interface: map-reduce is a very low-level programming model, so each new business question meant a developer writing, debugging and then maintaining a bespoke job. Those programs were hard to reuse, baked the physical layout of the files into code, and left the data unreachable for the analysts who knew SQL but not Java. Other map-reduce front ends existed, notably Pig and Microsoft's Scope, but they were dataflow languages without a system catalog, so schemas and storage formats lived inside individual scripts instead of a shared, queryable place. Hive was built to make a Hadoop cluster behave like the warehouse Facebook could no longer afford to buy.

The problem — what was actually breaking

Traditional warehousing solutions had become prohibitively expensive at the data volumes industry was collecting for business intelligence.
The map-reduce programming model is very low level and requires developers to write custom programs that are hard to maintain and reuse.
An analyst who could state a question in SQL could not run it against Hadoop data at all without a programmer translating it into map-reduce code.
Warehouse data arrives in many different serialization formats, so a system with one fixed IO library cannot read all the files it is asked to query.
Comparable map-reduce front ends such as Pig and Scope carry no system catalog, leaving nowhere to record schemas and statistics for data exploration and query optimization.
Running several different aggregations over the same input would naively rescan that input once per query, which dominates cost at warehouse scale.

Core ideas — the contributions, and why they work

HiveQL compiled to map-reduce

Hive exposes a SQL-like declarative language supporting select, project, join, aggregate, union all and sub-queries in the from clause, plus DDL to create tables with specific serialization formats and partitioning and bucketing columns, and DML load and insert statements. The compiler turns each statement into a plan, so the user states what they want and Hive decides how many map-reduce jobs it takes. This is the classic argument for declarative querying transplanted onto a batch execution engine: the physical decisions, such as where a shuffle boundary goes and which partitions to read, become the compiler's job rather than the analyst's. The immediate payoff at Facebook was that people who could write SQL stopped needing a Java programmer as an intermediary.

The Metastore as a real system catalog

Hive keeps a system catalog holding databases, tables and partitions, with columns and types, owner, storage location, bucketing information and the class names of the serializer and deserializer for that data. The paper is explicit that this is what distinguishes Hive as a traditional warehousing solution, in the manner of Oracle or DB2, from other map-reduce front ends like Pig and Scope. A catalog means the schema is declared once at table creation and reused on every reference, so type checking, select-star expansion and partition pruning are all possible at compile time. It also makes the warehouse browsable: a user can discover what tables exist and what they contain without opening a single file.

Tables, partitions and buckets as directory structure

A table is an HDFS directory, a partition is a subdirectory named by its column values, such as /wh/T/ds=20090101/ctry=US, and a bucket is a single file inside a partition directory chosen by hashing a column. The whole physical data model is therefore encoded in path names that both Hive and Hadoop can interpret without any index. This is why partition pruning is so cheap: eliminating a predicate range means not listing a directory, and it needs no auxiliary structure to maintain. Bucketing gives the same trick one level down, letting sampling queries skip whole files, and gives joins a pre-hashed layout to exploit.

SerDe: pluggable serialization, schema on read

Rather than requiring data to be loaded into a proprietary format, Hive associates each table with a serialization format and stores that association in the catalog; builtin formats exploit compression and lazy de-serialization, and users can add new formats by writing custom serialize and de-serialize methods in Java. Because the SerDe class is metadata rather than code inside a query, the compiler and the execution engine pick it up automatically at compile and run time. The consequence is that files can be registered as external tables, over HDFS, NFS or local directories, and queried where they already sit. This is the schema-on-read posture that separates a Hadoop warehouse from a loading-first relational warehouse.

Multi-table insert over a shared scan

A single HiveQL statement may contain a from clause followed by several insert clauses, so several different queries over the same input are expressed together. Hive optimizes this by sharing the scan of the input data, which in the paper's example means the expensive join of status updates against profiles happens exactly once while both the gender summary and the school summary are produced from it. On a system where reading the input is the dominant cost and every stage boundary means writing to HDFS, amortizing the scan across output tables is one of the largest wins available. It is effectively a hand-declared form of multi-query optimization, which the paper lists as future work in its general form.

Escape hatches for what SQL cannot say

HiveQL admits user defined column transformation and aggregation functions written in Java, and, more radically, lets users embed custom map-reduce scripts in any language through MAP and REDUCE clauses with a row-based streaming interface, reading rows from standard input and writing rows to standard output. The paper's own example uses a Python meme-extractor as a mapper and a Python top10.py as a reducer, precisely because Hive did not yet have a rank aggregation function. This admits that a young SQL dialect will not cover every need, and turns that gap into a plug-in point instead of a wall. The paper is candid that the flexibility costs the conversion of rows to and from strings.

How it works — the mechanism, concretely

Physical layout on HDFS

Each table has a corresponding HDFS directory, and the rows are serialized into files within it using the format recorded in the catalog. Partition columns are not stored in the data at all: they are encoded in the subdirectory path, so a table T partitioned on ds and ctry stores the rows with ds=20090101 and ctry=US in files under /wh/T/ds=20090101/ctry=US. Within a partition, data may be divided into buckets by the hash of a column, with each bucket materialized as one file in the partition directory. Column types may be primitives (integers, floating point numbers, generic strings, dates, booleans) or nestable collections, array and map, and users can define their own types programmatically.

The Metastore and why it is not on HDFS

The metastore holds Database objects as namespaces (the default database is named default), Table objects with columns and types, owner, storage information including data location, format and bucketing, plus SerDe implementation classes and arbitrary user key-value pairs intended to later carry statistics, and Partition objects that may carry their own columns, SerDe and storage information as a hook for future schema evolution. Its storage engine must be optimized for online transactions with random access and updates, which HDFS is not, being optimized for sequential scans, so the metastore runs on a relational database such as MySQL or Oracle, or on a plain file system such as local, NFS or AFS. The payoff is that HiveQL statements touching only metadata execute with very low latency. The cost is that data and metadata now live in two systems, and Hive has to explicitly maintain consistency between them.

Driver, Thrift server and the request path

External interfaces include a command line, a web UI, and JDBC and ODBC APIs, all of which reach Hive through a Thrift server that exposes a simple client API for executing HiveQL; because Thrift generates clients in many languages, the same Java server backs JDBC in Java, ODBC in C++, and scripting drivers in php, perl and python. The Driver receives the statement, creates a session handle used to track statistics such as execution time and number of output rows, and manages the statement's life cycle through compilation, optimization and execution. Once the compiler returns a DAG, the driver submits the individual map-reduce jobs to the execution engine in topological order. The execution engine is Hadoop, and every other component of Hive interacts with the metastore.

Compiler front end: parser and semantic analyzer

The parser turns the query string into a parse tree, and the semantic analyzer turns that parse tree into a block-based internal query representation. To do so it goes to the metastore for the schemas of the input tables, then verifies column names, expands select star into the actual column list, and does type checking, inserting implicit type conversions where needed. This is where catalog and query meet: without the metastore this stage would have nothing to check against. For DDL statements the resulting plan contains only metadata operations, and for LOAD statements only HDFS operations, so only inserts and queries go on to produce map-reduce work.

Logical plan and the rule-based optimizer

The logical plan generator converts the internal representation into a tree of logical operators, and the optimizer then makes multiple passes over it. It combines multiple joins that share a join key into a single multi-way join so they collapse into one map-reduce job; it inserts repartition operators, also called ReduceSinkOperators, in front of join, group-by and custom map-reduce operators to mark where a map phase must end and a reduce phase begin; it prunes columns early and pushes predicates down toward the table scans to shrink the data moving between operators; and it prunes unneeded partitions for partitioned tables and unneeded buckets for sampling queries. Users can additionally hint the optimizer to add partial aggregation operators for high-cardinality grouped aggregation, to add repartition operators to spread skew in grouped aggregation, or to perform a join in the map phase rather than the reduce phase.

Physical plan generation: cutting the tree into jobs

The physical plan generator turns the logical plan into a DAG of map-reduce jobs by creating a new job for each marker operator, namely repartition and union all, and assigning the slices of the logical plan lying between markers to the mappers and reducers of those jobs. Within one job, the part of the operator tree below the ReduceSinkOperator runs in the mapper and the part above it runs in the reducer, while the repartitioning itself is performed by the execution engine rather than by an operator. Figure 2 shows the multi-table insert example compiled into three jobs: the first performs the join and writes two temporary HDFS files, tmp1 and tmp2, which the second and third jobs consume to produce the gender and school summaries. That materialization is also the synchronization point, since the second and third jobs must wait for the first to finish.

What the paper showed — measurements and proofs

At the time of the paper the Facebook Hive warehouse contained several thousand tables and over 700 terabytes of data, used extensively for reporting and ad-hoc analysis by more than 100 users (Section 1).
The same warehouse instance supported over 5000 queries on a daily basis, with an active user and developer community inside and outside Facebook (Section 5).
In preliminary experiments running the benchmark of Pavlo et al. (SIGMOD 2009), the team improved the performance of Hadoop itself by 20 percent over the published numbers, largely by using faster Hadoop data structures, for example Text instead of String.
The same queries expressed in HiveQL carried a 20 percent overhead relative to that optimized hand-written Hadoop implementation, which the authors read as Hive being on par with the Hadoop code in the comparison study.
The multi-table insert example compiles to a DAG of exactly three map-reduce jobs (Figure 2), in which one join scan feeds two temporary HDFS files consumed by the two aggregation jobs.
HiveQL statements that access only metadata objects execute with very low latency, because the metastore is backed by a relational database or ordinary file system rather than HDFS (Section 3.1).

Limits and trade-offs — conceded and discovered

Conceded by the paper: HiveQL accepts only a subset of SQL as valid queries, with no support for updating or deleting rows in existing tables, and no rank aggregation function, which is exactly why the running example computes top-10 memes with an external Python reduce script.
Conceded by the paper: the optimizer is a naive rule-based one with a small number of simple rules, so there is no cost-based plan selection and users must supply hints to get map-side joins, partial aggregation or skew handling; a cost-based and adaptive optimizer is listed as future work.
Conceded by the paper: the streaming interface for custom map-reduce scripts requires converting every row to and from strings, and storage is row-oriented, with columnar storage and smarter data placement only being explored at the time.
Conceded by the paper: because the metastore deliberately lives outside HDFS on a transactional store, Hive must explicitly maintain consistency between metadata and the data files, a split that later caused real problems with concurrent writers and stale partition listings.
Exposed by later work: every stage boundary materializes to HDFS and downstream jobs simply wait for their predecessors, so even small queries take minutes, and the paper's only performance numbers are preliminary and self-reported; this latency floor is precisely what Tez, Spark, Impala and Presto were built to remove.

What it became — the systems that inherited it

Hive made SQL the default interface to big data, and almost every SQL-on-Hadoop engine that followed either copied its interfaces or plugged directly into them. The Hive Metastore in particular outlived Hive's own execution engine and became the de facto catalog standard: Spark SQL, Presto and Trino, Impala, Drill and Flink SQL all read table, partition and SerDe metadata from it, and AWS Glue Data Catalog is deliberately Hive-metastore API compatible. Hive-style partitioning, the directory naming convention of column=value, became the interchange layout for object-store data lakes, and the modern table formats Iceberg, Delta Lake and Hudi were designed largely to fix what this layout could not do: atomic multi-partition commits, file-level statistics and snapshot isolation that directory listing alone cannot provide. The SerDe idea generalized into the schema-on-read stance of the whole ecosystem, with columnar formats such as RCFile, ORC and Parquet arriving as the columnar storage the paper says it is exploring. Hive itself was rebuilt around the paper's admitted weaknesses, moving from map-reduce to Tez, adding vectorization, cost-based optimization through Calcite, ACID transactions and LLAP caching. The deeper inheritance is architectural: a declarative language, a shared catalog, and a pluggable execution engine underneath is now the standard shape of a data-lake query system.

In the paper’s words — verbatim

“However, the map-reduce programming model is very low level and requires developers to write custom programs which are hard to maintain and reuse.”

§1 Introduction

“The metastore distinguishes Hive as a traditional warehousing solution (ala Oracle or DB2) when compared with similar data processing systems built on top of map-reduce like architectures like Pig [7] and Scope [2].”

§3.1 Metastore

“Hive currently has a naïve rule-based optimizer with a small number of simple rules.”

§5 Future Work

Vocabulary — as this paper uses it

Table: A named relation whose data lives in one HDFS directory, serialized into files using a format recorded in the catalog. Hive also supports external tables over data already sitting in HDFS, NFS or local directories.
Partition: A subdivision of a table whose values are encoded in the directory path rather than stored in the rows, for example /wh/T/ds=20090101/ctry=US. Partitions determine data distribution within the table directory and let the optimizer prune whole directories.
Bucket: A further division of a partition based on the hash of a column, with each bucket stored as one file in the partition directory. Buckets let sampling queries skip files and give joins a pre-hashed layout.
SerDe: The pair of custom serialize and de-serialize methods, written in Java, that let Hive read and write a data format. The SerDe implementation class is stored in the catalog and applied automatically during query compilation and execution.
Metastore: Hive's system catalog of databases, tables and partitions, holding columns and types, owner, storage location, bucketing and SerDe information. It runs on a relational database or ordinary file system rather than HDFS because it needs random-access updates.
HiveQL: Hive's SQL-like declarative language, covering select, project, join, aggregate, union all, sub-queries in the from clause, DDL for tables with serialization and partitioning options, and load and insert DML. It is compiled to a plan rather than interpreted row by row.
Multi-table insert: A single HiveQL statement that runs several queries over the same input and inserts each result into a different table or partition. Hive optimizes it by sharing one scan of the input data across all the outputs.
Repartition operator (ReduceSinkOperator): The marker operator the optimizer inserts before joins, group-bys and custom map-reduce operators to denote a shuffle. It marks the boundary between the map phase and the reduce phase, and the physical plan generator starts a new map-reduce job at each such marker.
External table: A table whose data Hive queries in place, on HDFS, NFS or a local directory, rather than data loaded into the warehouse directory. It is the mechanism by which existing files become queryable without a copy.

On the timeline — where this sits in the story

Year2009

MilestoneHive

Problem solvedMapReduce was too low-level for analysts and BI users who wanted SQL-like access to Hadoop data.

How it solved itHive introduced a SQL-like warehouse layer over Hadoop datasets.

Why it matteredHelped SQL become the common interface for big-data systems.

DDIACh. 11: Batch Processing

View on the timeline