hive and shark

Hive and Shark

Amir H. [email protected]

Amirkabir University of Technology(Tehran Polytechnic)

Amir H. Payberah (Tehran Polytechnic) Hive and Shark 1393/8/19 1 / 45

Motivation

I MapReduce is hard to program.

I No schema, lack of query languages, e.g., SQL.


Solution

I Adding tables, columns, partitions, and a subset of SQL to unstruc-tured data.


Hive

I A system for managing and querying structured data built on topof Hadoop.

I Converts a query to a series of MapReduce phases.

I Initially developed by Facebook.

I Focuses on scalability and extensibility.


Scalability

I Massive scale out and fault tolerance capabilities on commodityhardware.

I Can handle petabytes of data.


Extensibility

I Data types: primitive types and complex types.

I User Defined Functions (UDF).

I Serializer/Deserializer: text, binary, JSON ...

I Storage: HDFS, Hbase, S3 ...


RDBMS vs. Hive

RDBMS Hive

Language SQL HiveQL

Update Capabilities INSERT, UPDATE, and DELETE INSERT OVERWRITE; no UPDATE or DELETE

OLAP Yes Yes

OLTP Yes No

Latency Sub-second Minutes or more

Indexes Any number of indexes No indexes, data is always scanned (in parallel)

Data size TBs PBs


RDBMS vs. Hive

RDBMS Hive

Language SQL HiveQL

Update Capabilities INSERT, UPDATE, and DELETE INSERT OVERWRITE; no UPDATE or DELETE

OLAP Yes Yes

OLTP Yes No

Latency Sub-second Minutes or more

Indexes Any number of indexes No indexes, data is always scanned (in parallel)

Data size TBs PBs

I Online Analytical Processing (OLAP): allows users to analyzedatabase information from multiple database systems at one time.

I Online Transaction Processing (OLTP): facilitates and managestransaction-oriented applications.


Hive Data Model

I Re-used from RDBMS:• Database: Set of Tables.• Table: Set of Rows that have the same schema (same columns).• Row: A single record; a set of columns.• Column: provides value and type for a single value.


Hive Data Model - Table

I Analogous to tables in relational databases.

I Each table has a corresponding HDFS directory.

I For example data for table customer is in the directory/db/customer.


Hive Data Model - Partition

I A coarse-grained partitioning of a table based on the value of acolumn, such as a date.

I Faster queries on slices of the data.

I If customer is partitioned on column country, then data with aparticular country value SE, will be stored in files within the directory/db/customer/country=SE.


Hive Data Model - Bucket

I Data in each partition may in turn be divided into buckets based onthe hash of a column in the table.

I For more efficient queries.

I If customer country partition is subdivided further into buckets,based on username (hashed on username), the data for each bucketwill be stored within the directories:/db/customer/country=SE/000000 0

...

/db/customer/country=SE/000000 5


Column Data Types

I Primitive types• integers, float, strings, dates and booleans

I Nestable collections• array and map

I User-defined types• Users can also define their own types programmatically


Hive Operations

I HiveQL: SQL-like query languages

I DDL operations (Data Definition Language)• Create, Alter, Drop

I DML operations (Data Manipulation Language)• Load and Insert (overwrite)• Does not support updating and deleting

I Query operations• Select, Filter, Join, Groupby


DDL Operations (1/3)

I Create tables

-- Creates a table with three columns

CREATE TABLE customer (id INT, name STRING, address STRING)

ROW FORMAT DELIMITED FIELDS TERMINATED BY ’\t’;

I Create tables with partitions

-- Creates a table with three columns and a partition column

-- /db/customer2/country=SE;

-- /db/customer2/country=IR;

CREATE TABLE customer2 (id INT, name STRING, address STRING)

PARTITION BY (country STRING)



I Create tables with buckets

-- Specify the columns to bucket on and the number of buckets

-- /db/customer3/000000_0



set hive.enforce.bucketing = true;

CREATE TABLE customer3 (id INT, name STRING, address STRING)

CLUSTERED BY (id) INTO 3 BUCKETS;

I Browsing through tables

-- lists all the tables

SHOW TABLES;

-- shows the list of columns

DESCRIBE customer;



I Altering tables

-- rename the customer table to alaki

ALTER TABLE customer RENAME TO alaki;

-- add two new columns to the customer table

ALTER TABLE customer ADD COLUMNS (job STRING);

ALTER TABLE customer ADD COLUMNS (grade INT COMMENT ’some comment’);

I Dropping tables

DROP TABLE customer;


DML Operations

I Loading data from flat files.

-- if ’LOCAL’ is omitted then it looks for the file in HDFS.

-- the ’OVERWRITE’ signifies that existing data in the table is deleted.

-- if the ’OVERWRITE’ is omitted, data are appended to existing data sets.

LOAD DATA LOCAL INPATH ’data.txt’ OVERWRITE INTO TABLE customer;

-- loads data into different partitions

LOAD DATA LOCAL INPATH ’data1.txt’ OVERWRITE INTO TABLE customer2

PARTITION (country=’SE’);

LOAD DATA LOCAL INPATH ’data2.txt’ OVERWRITE INTO TABLE customer2

PARTITION (country=’IR’);

I Store the query results in tables

INSERT OVERWRITE TABLE customer SELECT * From old_customers;


Query Operations (1/3)

I Selects and filters

SELECT id FROM customer2 WHERE country=’SE’;

-- selects all rows from customer table into a local directory

INSERT OVERWRITE LOCAL DIRECTORY ’/tmp/hive-sample-out’ SELECT *

FROM customer;

-- selects all rows from customer2 table into a directory in hdfs

INSERT OVERWRITE DIRECTORY ’/tmp/hdfs_ir’ SELECT * FROM customer2

WHERE country=’IR’;



I Aggregations and groups

SELECT MAX(id) FROM customer;

SELECT country, COUNT(*), SUM(id) FROM customer2 GROUP BY country;

INSERT TABLE high_id_customer SELECT c.name, COUNT(*) FROM customer c

WHERE c.id > 10 GROUP BY c.name;



I Join

CREATE TABLE customer (id INT, name STRING, address STRING)


CREATE TABLE order (id INT, cus_id INT, prod_id INT, price INT)


SELECT * FROM customer c JOIN order o ON (c.id = o.cus_id);

SELECT c.id, c.name, c.address, ce.exp FROM customer c JOIN

(SELECT cus_id, sum(price) AS exp FROM order GROUP BY cus_id) ce

ON (c.id = ce.cus_id) INSERT OVERWRITE TABLE order_customer;


User-Defined Function (UDF)

package com.example.hive.udf;

import org.apache.hadoop.hive.ql.exec.UDF;

import org.apache.hadoop.io.Text;

public final class Lower extends UDF {

public Text evaluate(final Text s) {

if (s == null) { return null; }

return new Text(s.toString().toLowerCase());

}

}

-- Register the class

CREATE FUNCTION my_lower AS ’com.example.hive.udf.Lower’;

-- Using the function

SELECT my_lower(title), sum(freq) FROM titles GROUP BY my_lower(title);


Executing SQL Questions

I Processes HiveQL statements and generates the execution planthrough three-phase processes.

1 Query parsing: transforms a query string to a parse tree representa-tion.

2 Logical plan generation: converts the internal query representationto a logical plan, and optimizes it.

3 Physical plan generation: split the optimized logical plan into multiplemap/reduce and HDFS tasks.


Optimization (1/2)

I Column pruning• Projecting out the needed columns.

I Predicate pushdown• Filtering rows early in the processing, by pushing down predicates to

the scan (if possible).

I Partition pruning• Pruning out files of partitions that do not satisfy the predicate.


Optimization (2/2)

I Map-side joins• The small tables are replicated in all the mappers and joined with

other tables.• No reducer needed.

I Join reordering• Only materialized and kept small tables in memory.• This ensures that the join operation does not exceed memory limits

on the reducer side.


Hive Components (1/8)



I External interfaces• User interfaces, e.g., CLI and web UI• Application programming interfaces, e.g., JDBC and ODBC• Thrift, a framework for cross-language services.



I Driver• Manages the life cycle of a HiveQL statement during compilation,

optimization and execution.



I Compiler (Parser/Query Optimizer)• Translates the HiveQL statement into a a logical plan, and

optimizes it.



I Physical plan• Transforms the logical plan into a DAG of Map/Reduce jobs.



I Execution engine• The driver submits the individual mapreduce jobs from the DAG to

the execution engine in a topological order.



I SerDe• Serializer/Deserializer allows Hive to read and write table rows in

any custom format.



I Metastore• The system catalog.• Contains metadata about the tables.• Metadata is specified during table creation and reused every time the

table is referenced in HiveQL.• Metadatas are stored on either a traditional relational database, e.g.,

MySQL, or file system and not HDFS.


Hive on Spark


Spark RDD - Reminder

I RDDs are immutable, partitioned collections that can be createdthrough various transformations, e.g., map, groupByKey, join.


Executing SQL over Spark RDDs

I Shark runs SQL queries over Spark using three-step process:

1 Query parsing: Shark uses Hive query compiler to parse the queryand generate a parse tree.

2 Logical plan generation: the tree is turned into a logical plan andbasic logical optimization is applied.

3 Physical plan generation: Shark applies additional optimization andcreates a physical plan consisting of transformations on RDDs.


Hive Components


Shark Components


Shark and Spark

I Shark extended RDD execution model:

• Partial DAG Execution (PDE): to re-optimize a running query afterrunning the first few stages of its task DAG.

• In-memory columnar storage and compression: to process relationaldata efficiently.

• Control over data partitioning.


Partial DAG Execution (1/2)

I How to optimize the following query?

SELECT * FROM table1 a JOIN table2 b ON (a.key = b.key)

WHERE my_crazy_udf(b.field1, b.field2) = true;

I It can not use cost-based optimization techniques that rely on ac-curate a priori data statistics.

I They require dynamic approaches to query optimization.

I Partial DAG Execution (PDE): dynamic alteration of query plansbased on data statistics collected at run-time.


Partial DAG Execution (2/2)

I The workers gather customizable statistics at global and per-partition granularities at run-time.

I Each worker sends the collected statistics to the master.

I The master aggregates the statistics and alters the query plan basedon such statistics.


Columnar Memory Store

I Simply caching Hive records as JVM objects is inefficient.

I 12 to 16 bytes of overhead per object in JVM implementation:• e.g., storing a 270MB table as JVM objects uses approximately 971

MB of memory.

I Shark employs column-oriented storage using arrays of primitive ob-jects.


Data Partitioning

I Shark allows co-partitioning two tables, which are frequently joinedtogether, on a common key for faster joins in subsequent queries.


Shark/Spark Integration

I Shark provides a simple API for programmers to convert results fromSQL queries into a special type of RDDs: sql2rdd.

val youngUsers = sql2rdd("SELECT * FROM users WHERE age < 20")

println(youngUsers.count)

val featureMatrix = youngUsers.map(extractFeatures(_))

kmeans(featureMatrix)


Summary

I Operators: DDL, DML, SQL

I Hive architecture vs. Shark architecture

I Add advance features to Spark, e.g., PDE, columnar memory store


Questions?


hive and shark

Technology

shark amir

hive data model re

structured data

example data

petabytes of data

extensibility data types

unstruc tured data

number of indexes