Improving Query Performance Using Partitioning & Bucketing in Apache Hive

To improve the Query performance we can go for

1)Partitioning tables
 
   a)Manual partition
   b)Dynamic partiton

2)Bucketing

Manual partition:

In Manual partition we are partitioning the table using partition variables. Here in our dataset we are
trying to partition by country and city names.

create table if not exists empl_part (empid int,ename string,salary double,deptno int)
comment 'manual partition example'
partitioned by (country string,city string)
row format delimited
fields terminated by ','


-- data for UK

load data local inpath 'uk_edinburgh.csv' into table empl_part partition(country='UK',city='EDIN');
load data local inpath 'uk_london.csv' into table empl_part partition(country='UK',city='LON');

-- data for usa

load data local inpath 'usa_newyork.csv' into table empl_part partition(country='US',city='NY');
load data local inpath 'usa_california.csv' into table empl_part partition(country='US',city='CF');



-- count(*)  -- count(1) where counrty and city
-- select * from employee order by empid asc;

-- to stop cli from querying the entire database

set hive.mapred.mode=strict;
--nonstrict  -- to make sure that all rows are not selected cos its may cause a massive mapreduce job

> select * from empl_part order by empid asc;
-- error FAILED: SemanticException [Error 10041]: No partition predicate found for Alias "employee" Table "employee"


show partitions empl_part; (Lists available partitions)

country=UK/city=EDIN
country=UK/city=LON
country=US/city=CF
country=US/city=NY

select * from empl_part where country='UK' order by empid asc;

When i tried using manual partition there is significant change in query performance compare to
querying on a normal table.


Dynamic partitioning

DP columns are specified the same way as it is for SP columns – in the partition clause. The only difference is that DP columns do not have values, while SP columns do. In the partition clause, we need to specify all partitioning columns, even if all of them are DP columns.
In INSERT ... SELECT ... queries, the dynamic partition columns must be specified last among the columns in the SELECT statement and in the same order in which they appear in the PARTITION() clause.

all DP columns – only allowed in nonstrict mode.

Process of dynamic partitioning

1) create a staging table
2) create the production table with partitions
3) load data from staging table into production table


CREATE TABLE IF NOT EXISTS EMPL_STAGE(EMPNO INT,ENAME STRING,SAL INT,DEPTNO INT,COUNTRY STRING,CITY STRING,DATE_JOINING STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ',';

LOAD DATA INPATH 'emp' into table EMPL_STAGE;

CREATE TABLE IF NOT EXISTS EMPL_PROD(EMPNO INT,ENAME STRING,SAL INT,COUNTRY STRING,CITY STRING,DATE_JOINING STRING)
PARTITIONED BY (DEPTNO INT)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ',';


-- set session properties
set  hive.exec.dynamic.partition=true;
set  hive.exec.dynamic.partition.mode=nonstrict;

INSERT OVERWRITE TABLE EMPL_PROD PARTITION(DEPTNO)
SELECT EMPNO,ENAME,SAL,COUNTRY,CITY,DATE_JOINING,DEPTNO as DEPTNO  FROM EMPL_STAGE

select count(*) from empl_stage where deptno=1;


Bucketing

When ever we do a select query on a table it has to go through whole table to retrieve the data.
which may become a performance issue and sometimes we may run out of memory.Here comes bucketing comes into the picture.

Hashing will be done on all the values internally and the values are dumped into buckets. As a result when ever we are firing a query the data will be fetched from a respective bucket.

Process of bucketing

1) create a staging table

2) create the production table with buckets

3) load data from staging table into production table

create external table nyse_daily_staging (

exchange string,

symbol string,

date_of_trading string,

open double,

high double,

low double,

close double,

volume int,

adj_close double)

row format delimited

fields terminated by '\t';

load data local inpath 'NYSE_daily' overwrite into table nyse_daily_staging;

create table nyse_daily_production (

exchange string,

symbol string,  

date_of_trading string,

open double,

high double,

low double,

close double,

volume int,

adj_close double)

clustered by (symbol) into 20 buckets

row format delimited

fields terminated by '\t';

--   assuming number of symbols are greater than the bucketsize

-- we need to enforce bucketing while loading data we set a hive parameter

set hive.enforce.bucketing=true ;

from nyse_daily_staging 

insert overwrite table nyse_daily_production

select * ;

tests

select * from nyse_daily_production where symbol = 'CBC' and open > 2.9;  160.599 sec

select * from nyse_daily_staging where symbol = 'CBC' and open > 2.9; 220.566 sec

Complex Data Types in HIVE

There are three complex types in hive.

Arrays: It is an ordered collection of elements.The elements in the array must be of the same type.

Map: It is an unordered collection of key-value pairs.Keys must be of primitive types.Values can be of any type.

Struct: It is a collection of elements of different types.


Examples:

Struct:

If the data pattern is like 100,John$Martin$Doe (customerID,Customer First middle and lastname seperated by '$')

create table cust_struct(
custid int,
name struct<fname:string,mname:string,lname:string>
)
row format delimited
fields terminated by ','
collection items terminated by '$';

Loading Data

!hdfs dfs -copyFromLocal cust-struct.dat /user/hive/warehouse/cust_struct;

Extracting data
select name.fname from cust_struct;


Map:

Data pattern:

100, John$Martin$Doe, home#01234$office#00000$mobile#9999

(last field is a map but map is also a collection terminated by '$' to identify the various key-value pairs )

create table cust_struct_map (
custid int,
name struct<fname:string,mname:string,lname:string>,
phone_nos map<string,int>
)
row format delimited
fields terminated by ','
collection items terminated by '$'
map keys terminated by '#';

Loading Data

!hdfs dfs -copyFromLocal cust-struct-map.dat /user/hive/warehouse/cust_struct_map;

Extracting data:

select name.fname,phone_nos["home"] from cust_struct_map;


Arrays:

Data pattern:

100,John$Martin$Doe,home#01234$office#00000$mobile#9999,abc@yahoo.com$doe@ooo.com

create table cust_struct_map_array(
custid int,
name struct<fname:string,mname:string,lname:string>,
phone_nos map<string,int>,
emails array<string>
)
row format delimited
fields terminated by ','
collection items terminated by '$'
map keys terminated by '#';

Loading Data

!hdfs dfs -copyFromLocal cust-struct-map-array.dat /user/hive/warehouse/cust_struct_map_array;

Extracting Data

select custid,emails from cust_struct_map_array;

Overview of IMPALA | IMPALA vs PIG & HIVE

What is IMPALA ?

Impala is a High-performance SQL Engine for vast amounts of data.

• Massively parallel procesing
• Inspired by google's Dremel project
• Query latency measured in milliseconds

Impala runs on hadoop clusters

• Can query data store in HDFS or HBase tables
• Read and Write data in common hadoop file formats

Developed by cloudera

• 100% open source under apache server licence

Impala supports a subset of SQL-92

Why to use IMPALA ?

Many benefits are the same as with Hive or PIG

• More productive than writing Map-Reduce code 
• Leverage existing knowledge of SQL

One benefit exclusive to impala is Speed

• Highly optimized for queries
• Almost 5times faster than Hive or Pig. Often 20 times faster or more

IMPORTANT:

Pig & Hive functions from your laptop where as the impala sits on Hadoop cluster on all data nodes.

Comparing Impala to Hive & Pig

Similarities:

• Queries expressed in high-level languages
• Alternatives to writting mapreduce code
• Used to analyze data stored on Hadoop clusters
• Impala shares the meta store with Hive (Tables created in Hive as visible in Impala (viceversa))

Contrasting Impala to Hive & Pig

• Hive & Pig answers queries by running Mapreduce jobs.Map reduce over heads results in high latency.(even a trivial query takes 10sec or more)
• Impala does not use mapreduce.It uses a custom execution engine build specifically for Impala.
• Queries can complete in a fraction of sec.
• Hive & Pig are best suited long-running batch processes (Data Transformation Tasks).Impala best for interactive/Adhoc queries.
• Impala can't handle complex data types(Array,Map or Struct)
• No support for binary data type.

If a node fails in Hive or Pig, They answers queries by running mapreduce jobs in other nodes. But,incase of impala if the node fails during a query,the query will fails and it has to be re-run

Relation Database vs Impala

How can we import MySQL data into HDFS | Working with Sqoop

Sqoop:

• Sqoop is a tool that used to import SQL data from SQL databases to HDFS. It was devloped by cloudera.
• It uses JDBC to talk to database.
• Sqoop provides the java code that can import data to HDFS
• After the java code generation,a map only mapreduce job is run to import the data.
• By default 4mappers are run with 25% each.

Here i am providing few commands that can list the databases,tables and import the tables.

1: sqoop list-databases --connect jdbc:mysql://localhost/training_db --username root --password root
(This will list the available databases in your connection)

2:sqoop list-tables --connect jdbc:mysql://localhost/training_db --username root --password root

3:sqoop list-tables --connect jdbc:mysql://localhost/training_db --table user_log --fields-terminated-by '\t'  -m 1 --username root --password root