Map side join in Hive
Category : Hive
Many time we face a situation that we have very small tables in hive but when we query these tables then it takes long time.
Here I am going to explain Map side join and its advantages over the normal join operation in Hive. But before knowing about this, we should first understand the concept of ‘Join’ and what happens internally when we perform the join in Hive.
Join is a clause that combines the records of two tables (or Data-Sets).
Assume that we have two tables A and B. When we perform join operation on them, it will return the records which are the combination of all columns o f A and B.
Mapjoin is a little-known feature of Hive. It allows a table to be loaded into memory so that a (very fast) join could be performed entirely within a mapper without having to use a Map/Reduce step. If your queries frequently rely on small table joins (e.g. cities or countries, etc.) you might see a very substantial speed-up from using mapjoins.
There are two ways to enable it. First is by using a hint, which looks like
/*+ MAPJOIN(aliasname), MAPJOIN(anothertable) */. This C-style comment should be placed immediately following the
SELECT. It directs Hive to load
aliasname (which is a table or alias of the query) into memory.
SELECT /*+ MAPJOIN(c) */ * FROM orders o JOIN cities c ON (o.city_id = c.id);
Another (better, in my opinion) way to turn on mapjoins is to let Hive do it automatically. Simply set
hive.auto.convert.join to true in your config, and Hive will automatically use mapjoins for any tables smaller than
hive.mapjoin.smalltable.filesize (default is 25MB).
Assume that we have two tables of which one of them is a small table. When we submit a map reduce task, a Map Reduce local task will be created before the original join Map Reduce task which will read data of the small table from HDFS and store it into an in-memory hash table. After reading, it serializes the in-memory hash table into a hash table file.
In the next stage, when the original join Map Reduce task is running, it moves the data in the hash table file to the Hadoop distributed cache, which populates these files to each mapper’s local disk. So all the mappers can load this persistent hash table file back into the memory and do the join work as before. The execution flow of the optimized map join is shown in the figure below. After optimization, the small table needs to be read just once. Also if multiple mappers are running on the same machine, the distributed cache only needs to push one copy of the hash table file to this machine.
Create two sample table and and insert some data into those table and then perform map join operation.
hive> create table emp(name string,address string, deptid bigint) row format delimited fields terminated by ‘,’;
Time taken: 20.218 seconds
hive> load data local inpath ‘/root/emp.txt’ overwrite into table emp;
Loading data to table test.emp
Table test.emp stats: [numFiles=1, numRows=0, totalSize=56, rawDataSize=0]
Time taken: 0.713 seconds
hive> select * from emp;
Saurabh AA 1
Babu AA 2
Nach BB 2
Jeba CC 1
Abhijit DD 1
Time taken: 20.105 seconds, Fetched: 5 row(s)
hive> create table dept(deptname string, deptid bigint) row format delimited fields terminated by ‘,’;
Time taken: 20.192 seconds
hive> load data local inpath ‘/root/dept.txt’ overwrite into table dept;
Loading data to table test.dept
Table test.dept stats: [numFiles=1, numRows=0, totalSize=13, rawDataSize=0]
Time taken: 20.705 seconds
hive> select * from dept;
Time taken: 0.081 seconds, Fetched: 2 row(s)
Without Map join:
hive> select emp.name,dept.deptname from emp join dept on emp.deptid=dept.deptid;
Query ID = root_20161018080320_198dcd9e-7e47-440f-871d-5da56522fced
Total jobs = 1
Launching Job 1 out of 1
Number of reduce tasks not specified. Estimated from input data size: 1
In order to change the average load for a reducer (in bytes):
In order to limit the maximum number of reducers:
In order to set a constant number of reducers:
Starting Job = job_1476772397810_0003, Tracking URL = http://m1.hdp22:8088/proxy/application_1476772397810_0003/
Kill Command = /usr/hdp/18.104.22.168-2557/hadoop/bin/hadoop job -kill job_1476772397810_0003
Hadoop job information for Stage-1: number of mappers: 2; number of reducers: 1
2016-10-18 08:04:53,713 Stage-1 map = 0%, reduce = 0%
2016-10-18 08:05:52,017 Stage-1 map = 100%, reduce = 0%, Cumulative CPU 3.08 sec
2016-10-18 08:06:39,906 Stage-1 map = 100%, reduce = 100%, Cumulative CPU 4.37 sec
MapReduce Total cumulative CPU time: 4 seconds 370 msec
Ended Job = job_1476772397810_0003
MapReduce Jobs Launched:
Stage-Stage-1: Map: 2 Reduce: 1 Cumulative CPU: 4.37 sec HDFS Read: 13044 HDFS Write: 52 SUCCESS
Total MapReduce CPU Time Spent: 4 seconds 370 msec
Time taken: 201.293 seconds, Fetched: 5 row(s)
With Map Join :
hive> select /*+ MAPJOIN(dept) */ emp.name,dept.deptname from emp join dept on emp.deptid=dept.deptid;
Query ID = root_20161018075509_476aa0ce-704a-4e3b-91c5-c2a3444a9fd7
Total jobs = 1
WARNING: Use “yarn jar” to launch YARN applications.
Execution log at: /tmp/root/root_20161018075509_476aa0ce-704a-4e3b-91c5-c2a3444a9fd7.log
2016-10-18 07:55:37 Starting to launch local task to process map join; maximum memory = 1065484288
2016-10-18 07:55:59 Dump the side-table for tag: 1 with group count: 2 into file: file:/tmp/root/f9baf1b2-42f4-4f89-9ac8-a48f5e8b0170/hive_2016-10-18_07-55-09_410_1173820569043720345-1/-local-10003/HashTable-Stage-3/MapJoin-mapfile11–.hashtable
2016-10-18 07:55:59 Uploaded 1 File to: file:/tmp/root/f9baf1b2-42f4-4f89-9ac8-a48f5e8b0170/hive_2016-10-18_07-55-09_410_1173820569043720345-1/-local-10003/HashTable-Stage-3/MapJoin-mapfile11–.hashtable (307 bytes)
2016-10-18 07:55:59 End of local task; Time Taken: 22.621 sec.
Execution completed successfully
MapredLocal task succeeded
Launching Job 1 out of 1
Number of reduce tasks is set to 0 since there’s no reduce operator
Starting Job = job_1476772397810_0002, Tracking URL = http://m1.hdp22:8088/proxy/application_1476772397810_0002/
Kill Command = /usr/hdp/22.214.171.124-2557/hadoop/bin/hadoop job -kill job_1476772397810_0002
Hadoop job information for Stage-3: number of mappers: 1; number of reducers: 0
2016-10-18 07:57:09,279 Stage-3 map = 0%, reduce = 0%
2016-10-18 07:57:57,324 Stage-3 map = 100%, reduce = 0%, Cumulative CPU 1.6 sec
MapReduce Total cumulative CPU time: 1 seconds 600 msec
Ended Job = job_1476772397810_0002
MapReduce Jobs Launched:
Stage-Stage-3: Map: 1 Cumulative CPU: 1.6 sec HDFS Read: 6415 HDFS Write: 52 SUCCESS
Total MapReduce CPU Time Spent: 1 seconds 600 msec
Time taken: 169.01 seconds, Fetched: 5 row(s)
I hope it will have help you to understand map join.