How to migrate a hive table from one hive instance to another or between hive databases

Hive has the EXPORT IMPORT feature since hive 0.8. With this feature you can export the metadata as well as the data for a corresponding table to a file in hdfs using the EXPORT command. The data is stored in json format. Data once exported this way could be imported back to another database or hive instance using the IMPORT command.

The syntax looks something like this:
EXPORT TABLE table_or_partition TO hdfs_path;
IMPORT [[EXTERNAL] TABLE table_or_partition] FROM hdfs_path [LOCATION [table_location]];

Some sample statements would look like:

EXPORT TABLE <table name> TO 'location in hdfs';

Use test_db;

IMPORT FROM 'location in hdfs';

Export Import can be appled on a partition basis as well:

EXPORT TABLE <table name> PARTITION (loc="USA") to 'location in hdfs';

The below import commands imports to an external table instead of a managed one

IMPORT EXTERNAL TABLE FROM 'location in hdfs' LOCATION ‘/location/of/external/table’;

Implementing basic SQL Update statement in Hive

Hive is not meant for point to point queries and hence sql update functionality would be least required in hive that should be the reason hive doesn’t have update functionality for rows or rather individual columns in a row. There would be cases you find a much more suitable use case in hive, but the same can’t be implemented as it includes an update statement on a few rows specified by a condition. We can implement the basic sql row update in hive by following a few series of steps as follows

1.       Load all the rows that has to be updated into a hdfs dir

a.       Here in the select statement that fetches the rows that satisfy the condition, we specify the column names one by one rather than a Select *

b.      In place of the colum whose value is to be updated we include the new value hard coded

2.       Remove these rows loaded into hdfs dir in step 1 from table

a.       Do an insert overwrite on the same table to include only those record that do not satisfies the condition

3.       Load the updated rows from hdfs dir in Step into into table

Let’s look at the same through an example

                We have a hive table employee with the following columns employeeId, employeeName, experienceMonths, salary and visaEligibility. We need to do an operation equivalent to the SQL query below in hive.

SQL Query

update table employee set visaEligibility = 'YES' where experienceMonths >36 ;

Equivalent Hive QL

#load the rows that fall in the condition into an hdfs dir with the corresponding column value modified

INSERT OVERWRITE DIRECTORY '/userdata/bejoy/employeetemp' SELECT employeeId,employeeName, experienceMonths ,salary,'YES' FROM employee WHERE experienceMonths > =36;

#removes the records from table that has been loaded into hdfs dir

#use the negate condition to the one provided in previous step

INSERT OVERWRITE TABLE employee SELECT * FROM employee WHEREexperienceMonths < 36;

#load the rows from hdfs dir on to table on top of its current contents

LOAD DATA INPATH '/userdata/bejoy/employeetemp' INTO TABLE employee;

A better Approach

                When I understood some more functionalities of hive, it was evident to me that there is a better way to solve this out using CASE statement. The same result could be obtained in a single hive query as stated below

INSERT OVERWRITE TABLE employee SELECT employeeId,employeeName, experienceMonths ,salary, CASE WHEN experienceMonths >=36 THEN ‘YES’ ELSE visaEligibility END AS visaEligibility FROM employee;

The approaches and solutions described in this post are for non partitioned tables. Ideally in real rime scenarios the data volume would be too large to handle in a single partition so we have to go in for multiple partitions. But the same approaches could be used out there as well.

Extract date in required formats from hive tables

Problem: My hive table has date in the format ‘yyyy-MM-dd HH:mm:ss’ which was obtained from  a database table through SQOOP. I need it in my query of the format ‘yyyy-MM-dd ’.

Solution: Use a regex expression to extract the required value.

When we sqoop in the date value to hive from rdbms, the data type hive uses to store that date is String. You can use a substring functions to achieve the same, the most easiest way would be to use the regexp_extract() function in hive.

To get the date alone in ‘yyyy-MM-dd ’ format we can go in for

regexp_extract(column_datetime,'(.*\-.*\-.*)\\s(.*)',1)

You can use it as Select *, regexp_extract(column_datetime,'(.*\-.*\-.*)\\s(.*)',1) from sample_table;

**column_datetime is a column of type string that stores date in the format ‘yyyy-MM-dd HH:mm:ss’ which is part of hive table sample_table

Some possible cases

Case: 01

If you want the timestamp alone regexp_extract(column_datetime,'(.*)\\s(.*)',2)

Case: 02

If you want the year and month alone of the format ‘yyyy-MM’ then useregexp_extract(column_datetime,'(.*\-.*)\-.*',1)

Case: 03

All these were mere String operations done using regular expressions now if I want my date in the format ‘dd-MM- yyyy’ it’d be better to go in for a combination date functions and string concatenation functions. We can achieve our goal using this combination of functions

concat(year(column_datetime),'-',month(column_datetime),'-',day(column_datetime) )

So we can use the same in our query  like

Select *, concat(day(column_datetime),'-',month(column_datetime),'-', year(column_datetime))

 from sample_table;

But there would be issues in using this that if dates and month values are less than 10 it would be represented with a single character, we can overcome the same with the use of case functionality as

concat(CASE WHEN day(install_datetime) < 10 THEN concat('0',day(install_datetime)) ELSE trim(day(install_datetime)) END,'-',CASE WHEN month(install_datetime) < 10 THEN concat('0',month(install_datetime)) ELSE trim(month(install_datetime)) END,'-',year(install_datetime))

A sample usage in hive QL would be as

select *,concat(CASE WHEN day(install_datetime) < 10 THEN concat('0',day(install_datetime)) ELSE trim(day(install_datetime)) END,'-',CASE WHEN month(install_datetime) < 10 THEN concat('0',month(install_datetime)) ELSE trim(month(install_datetime)) END,'-',year(install_datetime)) from sample_table;

How to create patritions on the fly in hive tables based on data? OR How can you implement Dynamic Partitions in hive for Larger Tables?

Straight answer depends on your business case but still partitions would be a better approach. Let us look at a business scenario on this issue

Problem: I have a huge table in my EDW that holds 5 billion rows. Every record has a column pay_location which has 300 distinct values across the table. Need to do some processing on the same within Hadoop environment and at a time my processing involves data only from certain pay_locations.

Table Schema in DWH

Invoice_Details
Column Name	Data Type
Invoice_Id	double
Invoice_Date	Date
Invoice_Amount	double
Pay_Location	VarChar
Paid_Date	Date

Solution: We can accomplish the same in 2 easy steps

Step 1: SQOOP import the table into hive from DWH

Step 2: Analyze the data using Hive QL

                Fist we need to SQOOP import the data into hive table ‘invoice_details’ using the basic SQOOP import command as follows.

sqoop import --driver <driver name> --connect <connection string> --username<username> -P --table Invoice_Details --split-by Invoice_Id --num-mappers <num of mappers> --warehouse-dir <hdfs dir> --hive-import --hive-table invoice_details_hive

We’d look into the second part here in more detail. How to effectively and efficiently analyze the same in hive. In our requirement it is clear that we can go ahead with a partitioned table approach for our data set, as the data analysis is made pay_location by pay_location we can do the partition based on pay_location itself.

Now how can we do the partition? Simple, need to create a partitioned table and load data into each partition. So first we can create an equivalent partitioned table in hive

CREATE TABLE invoice_details_hive _partitioned(Invoice_Id double, Invoice_Date string, Invoice_Amount double,Paid_Date string)PARTITIONED BY(pay_location string);

Once table creation is completed we need to load data into the partitions on invoice_details_hive _partitioned from invoice_details_hive . How can we do it? Can we go ahead for individual insert statements for each pay_location like?

INSERT OVERWRITE TABLE invoice_details_hive _partitioned PARTITION(pay_location=’USA’)

SELECT idh.Invoice_Id, idh.Invoice_Date, idh.Invoice_Amount, idh.Paid_Date FROM invoice_details_hive idh WHERE pay_location=’USA’;

If we follow this approach we may have to go in for 300 insert statements as there are 300 distinct values for pay_location in invoice_details_hive table. This type of implementation can be called as STATIC PARTIONS. But in our scenario static partitions won’t serve the purpose or rather it is too tedious. We’d have to implement the concept of DYNAMIC PARTITIONS introduced from hive 0.6 onwards. With Dynamic partitions we just need a single Insert Overwrite statement to create and load data into all the partitions.

INSERT OVERWRITE TABLE invoice_details_hive _partitioned PARTITION(pay_location)

SELECT idh.Invoice_Id, idh.Invoice_Date, idh.Invoice_Amount, idh.Paid_Date, idh.pay_location FROM invoice_details_hive idh;

This Single Query would implement dynamic partition for you, when you use dynamic partitions the last column from the select query on the source table should be column used for partitioning in the destination table (idh.pay_location)

When you try executing the query you can see hive throwing some fatal errors, like dynamic partition mode is strict and dynamic partition not enabled. So we need to set the following parameters in hive shell

1.       set hive.exec.dynamic.partition=true;

To enable dynamic partitions, by default it is false

2.       set hive.exec.dynamic.partition.mode=nonstrict;

We are using the dynamic partition without a static partition (A table can be partitioned based on multiple columns in hive) in such case we have to enable the non strict mode. In strict mode we can use dynamic partition only with a Static Partition.

3.       set hive.exec.max.dynamic.partitions.pernode=300;

The default value is 100, we have to modify the same according to the possible no of partitions that would come in your case

4.       set hive.exec.max.created.files=150000;

The default values is 100000 but for larger tables it can exceed the default, so we may have to update the same

In practical scenarios I did find the Dynamic Partition not working with the above query on really large tables and shooting a java print error after completion of first map process. This could be due to the larger number of files created on the first map process. However a slight modification of the job can help you overcome the same, group the records in your hive query on the map process and process them on the reduce side. ie use a map reduce process to achieve your goal rather than two map process. You can implement the same in your hive query itself with the usage of DISTRIBUTE BY, so the modified query would be

FROM invoice_details_hive idh

INSERT OVERWRITE TABLE invoice_details_hive _partitioned PARTITION(pay_location)

SELECT idh.Invoice_Id, idh.Invoice_Date, idh.Invoice_Amount, idh.Paid_Date, idh.pay_location

DISTRIBUTE BY pay_location;

With this approach you don’t need to overwrite the hive.exec.max.created.files parameter.

Include values during execution time in hive QL/ Dynamically substitute values in hive

When you play around with data warehousing it is very common to come across scenarios where you’d like to submit values at run time. In production environments when we have to enable a hive job we usually write our series of hive operations in HQL on a file and trigger it using the hive –f option from a shell script or some workflow management systems like oozie. Let’s have this discussion limited to triggering the hive job from shell as it is the basic one.

      Say I’m having a hive job called hive_job.hql, normally from a shell I’d trigger the hive job as

hive -f hive_job.hql

If I need to set some hive config parameters, say I need to enable compression in hive then I’d include the following arguments along as

hive -f hive_job.hql  -hiveconf hive.exec.compress.output=true -hiveconf mapred.output.compress=true -hiveconf mapred.output.compression.codec=com.hadoop.compression.lzo.LzopCodec

Now the final one, say my hive QL is doing some operations on date range and this date is varying/dynamic. This date is to be accepted from CLI each time. We can achieve the same with the following steps

1.       Pass the variables as config parameters

2.       Refer these config parameters in your hive query

                Let me make it more specific with a small example. I need to perform some operation on records in a table called ‘test_table’ which has a column/field named ‘creation_date’ .(ie I need to filter records based on creation_date). The date range for the operation is supplied at run time. It is achieved as

1.       Pass the variables as config parameters

                Here we need to pass two parameters, the start date and end date to get all the records within a specific date range

hive -f hive_job.hql  -hiveconf start_date=2011-01-01 –hiveconf end _date=2011-12-31

2.       Refer these config parameters in your hive query

                In our hive QL the start date and end date are to be decorated with place holders to be replaced by actual values during execution time.

SELECT * FROM test_table t1 WHERE t1.created_date=’ ${hiveconf: start_date }’ AND t2.created_date=’ ${hiveconf: end _date}’

Let us look into one more example, decide on the number of reducers during run time with the previous set of requirements along with compression. I have 2 components here 

·         Shell script that triggers .hql

Hive_trigger.sh

# The script accepts 3 parameters in order- number of reducers, start date and end date.

NUM_REDUCERS=$1

BEGIN_DATE=$2

CLOSE_DATE=$3

hive -f hive_job.hql  -hiveconf mapred.reduce.tasks= NUM_REDUCERS  -hiveconf start_date= BEGIN_DATE –hiveconf end _date= CLOSE_DATE

·         HQL that accepts parameters at run time

hive_job.hql

SELECT * FROM test_table t1 WHERE t1.created_date=’ ${hiveconf: start_date }’ AND t2.created_date=’ ${hiveconf: end _date}’

The components are ready you can now trigger you shell script. For example as
./hive_trigger.sh 50 2011-01-01 2011-12-31

Enable Sorted Bucketing in Hive

From the hive documents mostly we get to an impression as for grouping records, we go in for partitions and for sampling purposes, ie for evenly distributed records across multiple files we go in for buckets. But can we group records based on some columns/fields in buckets as well (individual files in buckets).

Concepts get clearer when we explain it through examples. So I’m taking the same route here.  Once with a hadoop assignment we did design a hadoop hybrid solution where the final output was on a hive partitioned table. This final output has to be consumed by an Oracle DWH for some legacy applications. The hand shake between hadoop and oracle team was they wanted ‘n’ files for each sub partition/folder and the files should have data grouped based on a few columns in the table (country and continent). If the files are grouped then the oracle load would be much efficient. How can we get the solution materialized?

1.       After hive operations do a map reduce on the final folders that would do the Group by

You do this by setting the number of reducers to ’n’ for n output files while running against each sub folder.It is really not a good solution because you have to run the map reduce for all sub partitions/folders which is definitely a performance glitch.

2.       Bucketing in hive

Using bucketing in hive for sub paritions. It is not plain bucketing but sorted bucketing. Normally we enable bucketing in hive during table creation as

CREATE EXTERNAL TABLE IF NOT EXISTS test_table

(

  Id INT,name String

)

PARTITIONED BY (dt STRING,hour STRING)

CLUSTERED BY(country,continent) INTO n BUCKETS

ROW FORMAT DELIMITED FIELDS TERMINATED BY '|'

LOCATION '/home/test_dir';

When we go into sorted bucketing/grouped bucketing our DDL would look like

CREATE EXTERNAL TABLE IF NOT EXISTS test_table

(         

Id INT,name String

)

PARTITIONED BY (dt STRING,hour STRING)

CLUSTERED BY(country,continent) SORTED BY(country,continent) INTO n BUCKETS

ROW FORMAT DELIMITED FIELDS TERMINATED BY '|'

LOCATION '/home/test_dir';     

Now to enforce bucketing while loading data into the table, we need to enable a hive parameter as follows

set hive.enforce.bucketing = true;

                     

With this DDL our requirement would be satisfied. The n individual files within each sub partitions and the records would be grouped into n files based on country, continent. ie the a particular combination of country, continent would be present in only one file. Now if the question arises, which combination in which file? It is decided by the hash partitioning function. If you want control over that you need to write your custom hash partitioner and plug in the same into your hive session.

NOTE: When we use partitions data is stored under individual directories/sub directories in hdfs. But when we use buckets the records are stored as files with naming convention ranging from 0 to n-1.

NOTE: In partitioned tables when we issue a query only the required partitions are scanned, no need to specify any hints in your hive query. But for bucketed tables it is not the case, you need to hint your hive query if you want to scan some particular buckets else the whole set of files would be scanned. We hint the buckets using TABLESAMPLE clause in our hive query. For example in our example if we want to choose only the data from BUCKET 2

SELECT * FROM test_table TABLESAMPLE(2 OUT OF n BUCKETS)WHERE dt=’2011-10-11’ AND hr=’13’;