The Hadoop ecosystem: HBase, Pig, Hive, Impala

HBase: a sorted map that looks like a table

In lesson 31 HDFS stored huge files that MapReduce reads whole, in batch. The deck that follows stacks four projects on it, one per complaint. Page 891 names three — "Pig – High-level language for data analysis", "HBase – Table storage for semi-structured data", "Hive – SQL-like Query language and Metastore" — and Impala comes later (Table 1).

Start with HBase. Page 893 describes it in five lines: "Modeled on Google's Bigtable", "Key-valued row/column store", "Billions of rows/millions on columns", "Column-oriented - nulls are free", "Untyped - stores byte[]". The example of the next slide explains each line.

A zoo in one row

A zoo keeps one row per enclosure (p. 894). The row key is enclosure1. Its columns are grouped into two column families: animal:, with the columns animal:type and animal:size, and repairs:, with repairs:cost. At time t1 the enclosure held a big lion; at time t2 a zebra moved in. The slide draws one row with two timestamp lines — t1: lion, big; t2: zebra — plus a repair cost of 1000 EUR (see the red pen below).

That is the whole data model: a map from cell addresses to bytes — in the Bigtable paper's words, "a sparse, distributed, persistent multidimensional sorted map" (background). The table of p. 894 is one way of drawing it.

What is written to disk

Page 895 shows the other way. Each column family is stored separately, as a list of entries, and each entry is a key (row, timestamp, column) with its value. Family animal: holds three entries — (enclosure1, t2, animal:type) zebra, (enclosure1, t1, animal:size) big, (enclosure1, t1, animal:type) lion. Family repairs: holds one, (enclosure1, t1, repairs:cost) 1000 EUR. Four entries in all.

The empty cells of the table have no entry. Nothing marks them and nothing reserves room for them: a missing entry simply is the null. That is "nulls are free". Space grows with the cells you write, not with rows times columns — why a store like this suits "semi-structured data" (p. 891), where every row fills a different handful of columns out of thousands. A relational table keeps a slot, or at least a NULL marker, for every declared column in every row (Fig. 1).

HBase stores one entry per written cell, (row, timestamp, column) → value, family by family; an empty cell is not stored at all.

Writing and reading by key

The code of p. 896 writes exactly this data, with the HBase API of the time:

// p. 896: lion and big at t1, then zebra at t2
HTable table = …
Text row = new Text("enclosure1");
Text col1 = new Text("animal:type");
Text col2 = new Text("animal:size");
BatchUpdate update = new BatchUpdate(row);
update.put(col1, "lion".getBytes("UTF-8"));
update.put(col2, "big".getBytes("UTF-8"));
table.commit(update);

update = new BatchUpdate(row);
update.put(col1, "zebra".getBytes("UTF-8"));
table.commit(update);

Read it aloud. A BatchUpdate collects the writes for one row; commit sends them together, so lion and big share a timestamp, t1. The second BatchUpdate writes only animal:type, at t2. Nothing is written to animal:size at t2: p. 894 leaves that cell empty, p. 895 has no entry for it, and a read of animal:size still returns big, its newest version. Every value goes in as bytes — "lion".getBytes("UTF-8") is "Untyped - stores byte[]" in code.

Page 897 lists the reads. There are three, and only three:

// p. 897: a cell (newest version: zebra), a row, a scan
Cell = table.getRow("enclosure1")
            .getColumn("animal:type").getValue();
RowResult = table.getRow( "enclosure1" );
Scanner s = table.getScanner( new String[] { "animal:type" } );

"Retrieve a cell", "Retrieve a row", "Scan through a range of rows": by key, by key, by a range of keys. Rows are kept sorted, so a range is one contiguous read. Nothing else: no SQL, no join, no secondary index — a condition on a value can only be checked while scanning (background, HBase guide). To find "all enclosures of the savannah", put the area into the row key so they sort together — the reasoning of Cassandra's partition key (lesson 21).

The slide's classes date from 2008; today they are Put, Get and Scan (background). Answer the exam with the slide's names. In today's HBase shell the zoo takes six lines:

# HBase shell today (background): the zoo of pp. 894–897
create 'zoo', {NAME => 'animal', VERSIONS => 3}, 'repairs'
put 'zoo', 'enclosure1', 'animal:type', 'lion'
put 'zoo', 'enclosure1', 'animal:size', 'big'
put 'zoo', 'enclosure1', 'animal:type', 'zebra'
get 'zoo', 'enclosure1', {COLUMN => 'animal:type', VERSIONS => 2}
scan 'zoo', {COLUMNS => 'animal:type'}

The get returns zebra, then lion. A family keeps only as many versions as it was created with — one by default since HBase 0.96 (background); the slide's example keeps two.

If the zoo reminds you of Cassandra (lesson 20), it should: both copy Bigtable's data model. Underneath, HBase keeps its files in HDFS and gives each range of rows to exactly one server, under a master, while Cassandra has no master (lesson 22 draws both). In the Hadoop stack HBase is the "online data access" of p. 835: random reads and writes by key, next to engines that read whole files.

Now write some cells yourself: every put adds exactly one entry to one family's list, and a get returns the newest version while the older ones stay stored.

Pig: nine lines, three MapReduce jobs

Page 899 introduces Pig: "Started at Yahoo! Research"; it "Expresses sequences of MapReduce jobs"; its "Data model: nested "bags" of items"; it "Provides relational (SQL) operators (JOIN, GROUP BY, etc.)"; and it is "Easy to plug in Java functions". The language is Pig Latin; Pig compiles a script into MapReduce jobs and runs them in order.

The question of p. 900: user data (name, age) in one file, website data (user, url) in another; find the top 5 most visited pages by users aged 18–25. The slide draws its plan as eight boxes: load users, filter by age, load pages, join on name, group on url, count clicks, order by clicks, take top 5.

Page 901 builds that plan in Java, every box by hand. To join, each mapper tags its records with the file they came from ("1" for pages, "2" for users), and a reducer splits them by the tag and forms every pair. To count, a mapper emits a 1 per visit and a reducer adds them up. To sort by clicks, a mapper swaps each pair so that the count becomes the key the framework sorts on. To keep the top, a single reducer stops after N records. Five jobs are chained by hand; each but the last writes its output to a temporary HDFS directory for the next one.

Page 902 says the same in nine lines of Pig Latin:

-- p. 902: top 5 pages of users aged 18-25
Users = load 'users' as (name, age);
Filtered = filter Users by age >= 18 and age <= 25;
Pages = load 'pages' as (user, url);
Joined = join Filtered by name, Pages by user;
Grouped = group Joined by url;
Summed = foreach Grouped generate group, COUNT(Joined) as clicks;
Sorted = order Summed by clicks desc;
Top5 = limit Sorted 5;
store Top5 into 'top5sites';

Read it line by line. Each line but the last names a new relation — a bag of tuples — built from earlier ones: load reads a file with a declared schema, filter keeps the users aged 18 to 25, join pairs each kept user with every visit whose user equals their name. Then the line that shows Pig's data model: group Joined by url counts nothing. It produces one tuple per url, (group, Joined), whose second field is the bag of all joined tuples with that url — the "nested bag" of p. 899. foreach … generate outputs, per group, the url and COUNT(Joined), the size of its bag: the clicks. Nothing runs until the store (background): Pig plans the whole flow at once.

In SQL the same question is a single statement (section 4 writes it in HiveQL); Pig spreads it over one step per line. "Easy to plug in Java functions" means user-defined functions, Java classes called by name inside foreach or filter (background).

Ease of translation: where the jobs come from

Page 903 lays the script next to the plan: each box is one statement, in order. Page 904, the final build of the same slide, frames the boxes into three MapReduce jobs: Job 1 holds load users, filter by age, load pages and join on name; Job 2 holds group on url and count clicks; Job 3 holds order by clicks and take top 5. Why these three?

1. A MapReduce job has at most one shuffle: map, then shuffle and sort, then reduce (lesson 30). Data is regrouped by a new key across the network once per job — or never, in a map-only job.

2. A join on name needs every kept user and every visit with the same name on one machine: a shuffle keyed by name. The two loads and the filter look at one tuple at a time, so they run in the map phase before it. Join on the reduce side: that is Job 1.

3. Grouping by url needs all joined tuples with the same url together: a shuffle keyed by url. It is a different key, so it needs a second job. Its map side reads Job 1's output from HDFS; counting each bag is its reduce side: Job 2.

4. Ordering by clicks needs the counts sorted across all urls: a shuffle keyed by clicks, a third job. Its reduce side reads them in order and keeps the first 5: Job 3.

5. So the number of jobs is the number of operators that need a shuffle — JOIN, GROUP, ORDER: three. Every other statement rides along in a map or a reduce phase of one of them.

Pig cuts a script into one MapReduce job per operator that needs a shuffle — JOIN, GROUP, ORDER — and folds every other statement into those jobs.

The Java of p. 901 used five jobs: its author wrote the two loads as separate map-only jobs. Pig needs three and writes fewer temporary files, but every job it keeps still costs a start-up and an output written to HDFS for the next job. Pig saves mostly your time, not the cluster's.

The widget runs the script on a toy dataset, one box at a time, and shows the jobs cut exactly at the shuffles. Its engine menu also runs the question in HiveQL and in Impala — come back to it after sections 4 and 5.

Hive: SQL and a Metastore over plain files

Page 906: Hive was "Developed at Facebook" and is "Used for majority of Facebook jobs". It is a "Relational database" built on Hadoop — with the quotation marks on the slide. It "Maintains list of table schemas", offers an "SQL-like query language (HiveQL)", "Can call Hadoop Streaming scripts from HiveQL" and "Supports table partitioning, clustering, complex data types, some optimizations".

The quotation marks are the point. Hive looks relational — tables, typed columns, SELECT, JOIN, GROUP BY — but has no storage or execution engine of its own: a query becomes MapReduce jobs over files in HDFS and takes as long as they take. In the Hive tutorial's words, "Hive is not designed for online transaction processing. It is best used for traditional data warehousing tasks" (background).

The architecture: metadata up here, data down there

Follow a query through p. 907 (Fig. 2). It arrives through the command line interface, the web interface, or the Thrift server, where programs connect with JDBC or ODBC as to any SQL database. The Driver compiles, optimizes and executes it: the compiler turns HiveQL into a plan of MapReduce jobs, looking the tables up in the Metastore — columns and types, partitions, and where in HDFS each table lives — and the executor submits the jobs to Hadoop, whose JobTracker runs their tasks next to the blocks.

Hive keeps the metadata — schemas, partitions, locations — in its Metastore; the data stays in HDFS as plain files, and its queries run as MapReduce jobs.

Tables, partitions, buckets

Page 908 lists the data model. Tables have "Typed columns (int, float, string, date, boolean)" and "Also, array/map/struct for JSON-like data": one column can hold a list, a dictionary or a record, which a flat SQL table cannot. Partitions split a table by the values of one or more columns — "e.g., to range-partition tables by date". Buckets are "Hash partitions within ranges (useful for sampling, join optimization)": inside a partition, the rows are spread over a fixed number of files by hashing one column — the "clustering" of p. 906.

Both uses follow from the hash. One bucket out of N is a sample of about one row in N, spread evenly over the values. And two tables bucketed on the same join key can be joined bucket by bucket, since equal keys always land in matching buckets (background: in HiveQL, CLUSTERED BY (userid) INTO 32 BUCKETS).

Storage: directories and flat files

Page 909 says where everything lives: a "Warehouse directory in HDFS – e.g., /user/hive/warehouse"; "Table row data stored in subdirectories of warehouse"; "Partitions form subdirectories of table directories"; "Actual data stored in flat files – Control char-delimited text, or SequenceFiles – With custom SerDe, can use arbitrary format".

So a table is a directory, a partition is a directory inside it, and the rows are ordinary files: text with control-character delimiters, or SequenceFiles, Hadoop's binary key-value format. A SerDe (serializer/deserializer) turns the bytes of a file into typed rows when a query reads them, and back when it writes. Hive checks nothing when files arrive: the Metastore's schema is applied at read time — the schema on read of lesson 4. And any other tool that reads HDFS can use the same files; Impala will. A table created with CREATE EXTERNAL TABLE … LOCATION '…' goes one step further: Hive registers files that stay where they are, and dropping the table deletes only its definition (background).

Page 910 creates the table that the rest of the deck queries:

-- p. 910: page views, one directory per (dt, country) pair
CREATE TABLE page_views(viewTime INT, userid BIGINT,
                        page_url STRING, referrer_url STRING,
                        ip STRING COMMENT 'User IP address')
COMMENT 'This is the page view table'
PARTITIONED BY(dt STRING, country STRING)
STORED AS SEQUENCEFILE;

Five typed columns, a comment on ip and one on the table. PARTITIONED BY declares dt and country, which are not in the column list: their values live in directory names, not in the files. STORED AS SEQUENCEFILE picks the file format. In the slide's words, "Partitioning breaks table into separate files for each (dt, country) pair".

The slide's example paths, "/hive/page_view/dt=2008-06-08,country=USA" and "…,country=CA", are a shorthand, with page_view for page_views. In HDFS each partition column is one directory level, under the warehouse of p. 909: /user/hive/warehouse/page_views/dt=2008-06-08/country=USA/, and next to it …/country=CA/ (Fig. 3 below draws the tree).

The Hive CLI of p. 913 lists and describes tables:

hive> show tables;
hive> describe page_views;
hive> describe extended page_views;

"List tables", "Describe a table", "More information": all three answer from the Metastore, without a job (background). describe prints the columns and their types, partition columns included; describe extended adds the HDFS location, the file format and the SerDe.

Asking Hive: partition pruning, joins and groups

Page 911 asks: "Find all page views coming from xyz.com on March 31st". As printed:

-- p. 911 as printed (see the red pen)
SELECT page_views.*
FROM page_views
WHERE page_views.date >= '2008-03-01'
AND page_views.date <= '2008-03-31'
AND page_views.referrer_url like '%xyz.com';

And the point of the slide: "Hive only reads partition 2008-03-01,* instead of scanning entire table". The mechanism has a name.

The two conditions of the query play opposite roles. The condition on the date is about a partition column — dt on p. 910, see the red pen — and its values are directory names, so the compiler can decide on it without reading a byte. The condition on referrer_url is about a column inside the files: it can only be checked row by row, by the mappers, on whatever survived the pruning. It reduces the output, never the input (Fig. 3).

Corrected for March 31st:

-- from xyz.com on March 31st: reads only dt=2008-03-31
SELECT page_views.*
FROM page_views
WHERE page_views.dt = '2008-03-31'
AND page_views.referrer_url LIKE '%xyz.com';

For the whole month, keep the slide's range on dt: 31 days of every country. The range works on a STRING column because dates written yyyy-mm-dd sort as strings in date order.

Partition pruning: a condition on a partition column chooses directories before any job runs; a condition on any other column filters rows inside the files that are read.

Joins and aggregation

Page 912: "Count users who visited each page by gender". As printed:

-- p. 912 as printed (see the red pen)
SELECT pv.page_url, u.gender, COUNT(DISTINCT u.id)
FROM page_views pv JOIN user u ON (pv.userid = u.id)
GROUP BY pv.page_url, u.gender
WHERE pv.date = '2008-03-03';

-- p. 912 corrected: distinct users per page and gender
SELECT pv.page_url, u.gender, COUNT(DISTINCT u.id)
FROM page_views pv JOIN user u ON (pv.userid = u.id)
WHERE pv.dt = '2008-03-03'
GROUP BY pv.page_url, u.gender;

COUNT(DISTINCT u.id) counts users, not views: a user who opened home.php ten times counts once — and there are tens of millions of them per page in one day. The plan has Pig's shape: a shuffle for the join on pv.userid = u.id, another for the grouping on (page_url, gender), and the jobs are cut at those shuffles. Pruning comes first: only the files of March 3rd enter the join.

The question of p. 900, asked in HiveQL over two tables users(name, age) and pages(username, url), is one statement — switch the engine of the widget above to HiveQL and step through it:

-- p. 900 in HiveQL: top 5 pages of users aged 18-25
SELECT p.url, COUNT(*) AS clicks
FROM users u JOIN pages p ON (u.name = p.username)
WHERE u.age >= 18 AND u.age <= 25
GROUP BY p.url
ORDER BY clicks DESC
LIMIT 5;

The same three shuffles — join, group, order — so the same three jobs, which Hive writes for you. Hive, like Pig, saves your time; the cluster still runs MapReduce.

Impala: SQL beside MapReduce, not on top of it

Page 915: Impala is a "Massive parallel processing (MPP) database engine, developed by Cloudera", "Integrated into Hadoop stack on the same level as MapReduce, and not above it (as Hive and Pig)". The figure of that slide is the thesis of the whole section (Fig. 4).

Why build it? Page 916 argues in six lines: "Data has a gravity"; "Today a lot of data live in HDFS"; "It is not practical to move big data"; "It is practical to bring engine to the data"; "In the same time – MapReduce is not must"; "Impala process data in Hadoop cluster without using MapReduce". Big data is too heavy to copy into a separate warehouse, so the engine moves onto the Hadoop machines — MapReduce's own "move the code, not the data" (lesson 30). But nothing forces that engine to be MapReduce, which starts every job from scratch and writes every output to disk. Nor is Impala alone: "Several other modern Database engines also realized the opportunity to bypass MapReduce but work right with HDFS" (p. 917, with Greenplum and JethroData).

The architecture: every node plans, coordinates and executes

Follow a query through p. 918 (Fig. 5). A client — any "SQL App", through ODBC, with "Common Hive SQL and interface" — sends it to one Impala node. That node's query planner reads the "Unified metadata": the Hive Metastore for the tables, the HDFS NameNode for where each block lives, the State Store for which nodes are alive. Its query coordinator cuts the plan into fragments and sends them to every node, "Fully MPP Distributed". Each node's query exec engine reads the blocks of its own DataNode — "Local Direct Reads" — or its HBase, and streams its rows back. No job is started, and nothing is written between the steps.

Same tables, faster reads

Page 919: "Impala supports multiple formats. It is kind of schema on read"; "Impala shares metastore with Hive, which enables very simple adoption"; "Internally Impala have well defined way to add new formats". The table page_views that Hive created is the table Impala queries: same definition in the Metastore, same directories, same files. You adopt Impala by pointing it at the warehouse you already have — data gravity again, nothing moves.

Page 920 lists what makes it fast. Impala's "format adapters", called scanners, "have predicate pushdown capability": the WHERE condition is handed to the code that reads the file, so a row that fails it is dropped during the read instead of being materialised and filtered later. It is an "Open source MPP engine" that "Runs hundreds of CPU cores in one query efficiently without expensive license" — and "Hive give us the same but not efficiently". There are scanners for the formats of p. 922 (Table 3).

With a columnar file, a scanner reads only the columns a query names — the column store of lesson 20, and the precise sense in which HBase is not column-oriented. Partition pruning skips directories, the columnar layout skips columns, predicate pushdown drops rows while they are read: together they make an analytic query read and process a small part of the data.

The price: one failure, one lost query

Page 921 is the honest slide: "Hive is doing things Impala can not do yet, like joins between several big tables"; "Hive has convenient java UDF, while impala has not"; "Impala does not have inter-query fault tolerance"; "In the same time – MapReduce is not good framework for the database engine"; "Always faster then Hive at least 10 times"; "Impala in the cloud is not elastic".

The other points follow from the same design. MPP joins build their tables in memory, so in the slide's version two big tables could not be joined, while Hive's reduce-side join spills to disk and handles any size, slowly. Java UDFs existed only in Hive, and Impala's daemons live on the DataNodes, so it cannot grow and shrink with demand like a cloud service. These describe the early Impala of the slides, and "at least 10 times" is the slide's own claim against Hive on MapReduce: learn them as printed, as facts about that version.

Back to the puzzle. Why keep an engine that is ten times slower? Because a long job must survive failures. A four-hour Hive job on hundreds of machines is likely to see some of them die, and MapReduce simply re-runs their tasks. An Impala query bets on being short enough to restart from scratch — a fine bet for a query of seconds, a ruinous one for a night of ETL. Hence the division of labour that follows from p. 921: Hive for long, heavy batch jobs and joins of big tables, Impala for interactive queries on the same tables, through the same Metastore. Speed was never the question; surviving failures, and tables too big for memory, was.

Impala is faster because it skips MapReduce — no job start-up, rows streamed in memory — and for the same reason a node failure kills the whole query: Hive for long fault-tolerant jobs, Impala for interactive ones.