Thursday, 30 June 2016

Handling skew data in Hive

Method 1: 
Identify the skew value and run the job for that value separately.
ex: If cust_id=100 has skew problem, then divide the records into cust_id=100 and cust_id!=100. then run the individual jobs.
Method 2:
Identify the column creating skew. If it is used for join, try to reduce the skew using multiple columns and use it in join.
Method 3:
Modify the join key(Salting). A simple approach that i follow some times is appending (key%3) at the end of key. ex: key1 can be divided into multiple values like key1_1,key1_2 etc,. This will help distribute the keys.
Method 4:
Divide the data into chunks and execute.

Sunday, 26 June 2016

Sqooping from/to MYSQL and HDFS

Sqoop Import
From MYSQL table to HDFS

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --target-dir '/user/horton/exported' -m 1

From MYSQL table to Hive

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported3 -m 1

From MYSQL table to Hive schema defined.

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported3 -m 1 --map-column-hive 'row_key=int,value=string,ts=timestamp';

From MYSQL table to Hive with delimiters

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported2 -m 1 --fields-terminated-by ','

Free form query import

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --query 'select * from exported where $CONDITIONS' --target-dir '/user/horton/test_query' -m 1

Incremental import into HDFS

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --target-dir '/user/horton/test_incremental' -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:16.0' -merge-key row_key

Incremental import into Hive

sqoop import --hive-import --connect jdbc:mysql://127.0.0.1/export --username hive --password hive --driver com.mysql.jdbc.Driver --table exported --hive-table test.exported1 -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:10.0' -merge-key row_key

Incremental import into HDFS from Query

sqoop import --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --query 'select * from exported where $CONDITIONS' --target-dir '/user/horton/test_incremental/t2' -m 1 --check-column ts --incremental lastmodified --last-value '2017-01-24 23:11:16.0' -merge-key row_key

Incremental import into Hive from Query is not supported.

Sqoop Export
Insert/export from HDFS

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --export-dir /apps/hive/warehouse/test.db/exported1  -m 1 --input-fields-terminated-by '\001'

Insert/export from Hive

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hcatalog-database test --hcatalog-table exported3 --input-fields-terminated-by '\001' -m 1

Update from HDFS

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --export-dir /apps/hive/warehouse/test.db/exported1  -m 1 --update-key row_key --input-fields-terminated-by '\001'

Using Hcatalog

sqoop export --connect jdbc:mysql://127.0.0.1/export --username root --driver com.mysql.jdbc.Driver --table exported --hcatalog-database test --hcatalog-table exported -m 1

Update from Hive using Hcatalog is not supported yet. Getting Java exception and there are open tickets on it.

Thursday, 18 February 2016

MapReduce V2 Job Flow

Mapreduce job flow on YARN involves below components.

  • A Client node, which submits the Mapreduce job.
  • The YARN Resource Manager, which allocates the cluster resources to jobs.
  • The YARN Node Managers, which launch and monitor the tasks of jobs.
  • The MapReduce Application Master, which coordinates the tasks running in the MapReduce job. The
  • application master and the MapReduce tasks run in containers that are scheduled by the resource
  • manager, and managed by the node managers.
  • The HDFS file system is used for sharing job files between the above entities.

Job Start up:

  • The call to Job.waitForCompletion() in the main driver class is where all the execution starts. The driver is the only piece of code that runs on our local machine, and this call starts the communication with the Resource Manager
  • Retrieves the new Job ID or Application ID from Resource Manager.
  • The Client Node copies Job Resources specified via the -files, -archives, and -libjars command-line arguments, as well as the job JAR file on to HDFS.
  • Finally, Job is submitted by calling submitApplication() method on Resource Manager.
  • Resource Manager triggers its sub-component Scheduler, which allocates containers for mapreduce job execution. Then Resource Manager starts Application Master in the container provided by the scheduler. This container will be managed by Node Manager from here on wards.

Input Split:


  • In this phase, HDFS splits the input files into equal sized chunks or segments based on minimum split size (mapreduce.input.fileinputformat.split.minsize) property .
  • Each file segment or split is passed to a unique map task if file is splittable. If File is not splittable then entire file will be provided as input to a single map task.
  • These map tasks are created by Mapreduce Application Master (MRAppMaster Java Class) and reduce tasks are also created by application master based on mapreduce.job.reduces property.

Role of an Application Master:


  • Before starting any task, Job setup method is called to create job’s output directory for job’s OutputCommitter.
  • As noted above, Both map tasks and reduce tasks are created by Application Master.
  • If the submitted job is small, then Application Master runs the job in the same JVM on which Application Master is running. It reduces the overhead of creating new container and running tasks in parallel. These small jobs are called as Uber tasks.
  • Uber tasks are decided by three configuration parameters, number of mappers <= 10, number of reducers <= 1 and Input file size is less than or equal to an HDFS block size. These parameters can be configured via mapreduce.job.ubertask.maxmaps , mapreduce.job.ubertask.maxreduces , and mapreduce.job.ubertask.maxbytes properties in mapred-site.xml.
  • If job doesn’t qualify as Uber task, Application Master requests containers for all map tasks and reduce tasks.

Task Execution:


  • Once Containers assigned to tasks, Application Master starts containers by notifying its Node Manager.
  • Node Manager copies Job resources (like job JAR file) from HDFS distributed cache and runs map or reduce tasks.
  • Running Tasks, keep reporting about the progress and status (Including counters) of current task to Application Master and Application Master collects this progress information from all tasks and aggregate values are propagated to Client Node or user.

Wednesday, 16 September 2015

To date calculations in SQL

CREATE TABLE m2dt(sale_amt INT, sale_dt DATE);
INSERT INTO m2dt VALUES(30,'2015-01-01'),(50,'2015-02-01'),(100,'2016-01-01'),(50,'2016-01-10'),(30,'2016-01-03');

Month to Date calculation:
SELECT 
SALE_DT,
SUM(sale_amt) OVER (PARTITION BY TRUNC(sale_dt,'MM') ORDER BY STRING(sale_dt)) AS amt 
FROM m2dt;

Year to Date calculation:
SELECT 
SALE_DT,
SUM(sale_amt) OVER (PARTITION BY YEAR(sale_dt) ORDER BY TRUNC(sale_dt,'MM')) AS amt 

FROM m2dt;


Friday, 15 May 2015

Hadoop Interview questions 2015 - Part3

1. How do you handle exceptions in MapReduce?
2. Modeling in NoSQL
3. Explain about one of the MapReduce program that you have implemented which u think it can't be done with Hive/Pig
4. What kind of initial validations that you will do on data before processing?
5. Explain when to choose MapReduce?
6. Differences b/w MapR 1 & 2
7. What is HA node in hadoop?  Do we have this in MapR 1?
8. What are the issues you faced in hadoop projects?
9.Difference between MapR and Cloudera distributions
10. What are the factors considered to decide key and the key format in mapreduce?

Sunday, 26 April 2015

Working of Hive

The following diagram depicts the workflow between Hive and Hadoop.
How Hive Works
The following table defines how Hive interacts with Hadoop framework:
Step No.Operation
1Execute Query
The Hive interface such as Command Line or Web UI sends query to Driver (any database driver such as JDBC, ODBC, etc.) to execute.
2Get Plan
The driver takes the help of query compiler that parses the query to check the syntax and query plan or the requirement of query.
3Get Metadata
The compiler sends metadata request to Metastore (any database).
4Send Metadata
Metastore sends metadata as a response to the compiler.
5Send Plan
The compiler checks the requirement and resends the plan to the driver. Up to here, the parsing and compiling of a query is complete.
6Execute Plan
The driver sends the execute plan to the execution engine.
7Execute Job
Internally, the process of execution job is a MapReduce job. The execution engine sends the job to JobTracker, which is in Name node and it assigns this job to TaskTracker, which is in Data node. Here, the query executes MapReduce job.
7.1Metadata Ops
Meanwhile in execution, the execution engine can execute metadata operations with Metastore.
8Fetch Result
The execution engine receives the results from Data nodes.
9Send Results
The execution engine sends those resultant values to the driver.
10Send Results
The driver sends the results to Hive Interfaces.

Saturday, 25 April 2015

How to Improve Hive Performance

Use Hive on Tez:

set hive.execution.engine=tez;
With the above setting, every HIVE query you execute will take advantage of Tez.

Use ORCFile:

Using ORCFile for every HIVE table should really be a no-brainer and extremely beneficial to get fast response times for your HIVE queries.

As an example, consider two large tables A and B (stored as text files, with some columns not all specified here), and a simple query like:

SELECT A.customerID, A.name, A.age, A.address join
B.role, B.department, B.salary
ON A.customerID=B.customerID;


This query may take a long time to execute since tables A and B are both stored as TEXT. Converting these tables to ORCFile format will usually reduce query time significantly:

CREATE TABLE A_ORC (
customerID int, name string, age int, address string
) STORED AS ORC tblproperties (“orc.compress" = “SNAPPY”);

INSERT INTO TABLE A_ORC SELECT * FROM A;

CREATE TABLE B_ORC (
customerID int, role string, salary float, department string
) STORED AS ORC tblproperties (“orc.compress" = “SNAPPY”);

INSERT INTO TABLE B_ORC SELECT * FROM B;

SELECT A_ORC.customerID, A_ORC.name,
A_ORC.age, A_ORC.address join
B_ORC.role, B_ORC.department, B_ORC.salary
ON A_ORC.customerID=B_ORC.customerID;



ORC supports compressed storage (with ZLIB or as shown above with SNAPPY) but also uncompressed storage.

Use Vectorization:

Vectorized query execution improves performance of operations like scans, aggregations, filters and joins, by performing them in batches of 1024 rows at once instead of single row each time.

Introduced in Hive 0.13, this feature significantly improves query execution time, and is easily enabled with two parameters settings:


set hive.vectorized.execution.enabled = true;
set hive.vectorized.execution.reduce.enabled = true;

Cost based query optimization:

Hive optimizes each query’s logical and physical execution plan before submitting for final execution. These optimizations are not based on the cost of the query – that is, until now.

A recent addition to Hive, Cost-based optimization, performs further optimizations based on query cost, resulting in potentially different decisions: how to order joins, which type of join to perform, degree of parallelism and others.

To use cost-based optimization (also known as CBO), set the following parameters at the beginning of your query:

set hive.cbo.enable=true;
set hive.compute.query.using.stats=true;
set hive.stats.fetch.column.stats=true;
set hive.stats.fetch.partition.stats=true;

Then, prepare the data for CBO by running Hive’s “analyze” command to collect various statistics on the tables for which we want to use CBO.

For example, in a table tweets we want to collect statistics about the table and about 2 columns: “sender” and “topic”:

analyze table tweets compute statistics;
analyze table tweets compute statistics for columns sender, topic;

With HIVE 0.14 (on HDP 2.2) the analyze command works much faster, and you don’t need to specify each column, so you can just issue:

analyze table tweets compute statistics for columns;

That’s it. Now executing a query using this table should result in a different execution plan that is faster because of the cost calculation and different execution plan created by Hive.


The Mindset Behind Reliable Data Systems I’ve been in data engineering long enough to see the stack change many times over. Tools come and g...