Hive Bucketed Map Join

Question

I am facing issue in executing bucketed map join.

I am using hive 0.10.

Table1 is a partitioned table on year,month and day. Each partition data is bucketed by column c1 into 128 buckets. I have almost 100 million records per day.

Table 1 
create table1
(
....
....
)
partitioned by (year int,month int,day int)
CLUSTERED BY(c1) INTO 128 BUCKETS;

Table2 is a large lookup table bucketed on column c1. I have 80 million records loaded into 128 buckets.

Table 2
create table2
(
 c1
 c2
 ...
)
CLUSTERED BY(c1) INTO 128 BUCKETS;

I have checked the data and it's loaded as per expectation into buckets.

Now, I am trying to enforce bucketed map join.That's where I am stuck.

set hive.auto.convert.join=true;
set hive.optimize.bucketmapjoin = true;
set hive.mapjoin.bucket.cache.size=1000000;

select a.c1 as c1_tb2,a.c2
       b.c1,b....
from table2 a
JOIN table1 b
ON (a.c1=b.c1);

I am still not getting bucketed map join. Am I missing something? Even I tried to execute join on only 1 partition. But, still I am getting same result.

Or

Bucketed map join doesn't work partition tables?

Please help.Thanks.

Answer 1

This explanation is for Hive 0.13. AFAICT, bucketed map join doesn't take effect for auto converted map joins. You will need to explicitly call out map join in the syntax like this:

set hive.optimize.bucketmapjoin = true;                                                   
explain extended select /* +MAPJOIN(b) */ count(*) 
from nation_b1 a 
join nation_b2 b on (a.n_regionkey = b.n_regionkey);

Note that only explain extended shows you the flag that indicates if bucket map join is being used or not. Look for this line in the plan.

BucketMapJoin: true

Answer 2

Tables are bucketed in hive to manage/process the portion of data individually. It will make the process easy to manage and efficient in terms of performance. Lets understand the join when the data is stored in buckets:

Lets say there are two tables user and user_visits and both table data is bucketed using user_id in 4 buckets . It means bucket 1 of user will contain rows with same user ids as that of bucket 1 of user_visits. And if a join is performed on these two tables on user_id columns, if it is possible to send bucket 1 of both tables to same mapper then good amount of optimization can be achieved. This is exactly done in bucketed map join.

Prerequisites for bucket map join:

Tables being joined are bucketized on the join columns, The number of buckets in one table is a same/multiple of the number of buckets in the other table. The buckets can be joined with each other, If the tables being joined are bucketized on the join columns. If table A has 4 buckets and table B has 4 buckets, the following join

SELECT /*+ MAPJOIN(b) */ a.key, a.valueFROM a JOIN b ON a.key = b.key

can be done on the mapper only. Instead of fetching B completely for each mapper of A, only the required buckets are fetched. For the query above, the mapper processing bucket 1 for A will only fetch bucket 1 of B. It is not the default behavior, and is governed by the following parameter

set hive.optimize.bucketmapjoin = true

If the tables being joined are sorted and bucketized on the join columns, and they have the same number of buckets, a sort-merge join can be performed. The corresponding buckets are joined with each other at the mapper. If both A and B have 4 buckets,

SELECT /*+ MAPJOIN(b) */ a.key, a.valueFROM A a JOIN B b ON a.key = b.key

can be done on the mapper only. The mapper for the bucket for A will traverse the corresponding bucket for B. This is not the default behavior, and the following parameters need to be set:

set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;