Limiting Joins in Apache Hive
Join the DZone community and get the full member experience.
Join For FreeThis article is by Stephen Mouring Jr, appearing courtesy of Scott Leberknight.
Working with large datasets in Hadoop / Hive works is difficult when you have an "imbalanced" join. If you are trying to join two tables together where a large amount of data in one table joins to just a few keys in the other table, individual reducers will get overloaded and slow your job down to the point where it may never finish. This is referred to in Hive as a skew join.
There are several ways to tackle this problem. If you have the ability to improve the quality of your data and fill in the blank values that is awesome, but that seldom is the case in the real world. You could also try to modify the join condition to include additional constraints that cause the data to be better partitioned, but again that is not likely to be the case in your problem domain.
Hive does claim to provide support some support for handling skew joins where it automatically divides up the data landing on the same key, runs it on separate reducers, then merges the results... Good luck with that.
So what are your alternatives?
A common cause of a skew join is when your data has a lot of null or blank keys. Whenever you see all the reducers except one complete very quickly you are often in this situation. You can verify that it is a null or blank key if the slow reducer has a reducer id of 0.
Map Reduce hashes each key to a number to determine which reducer it should land on (and ensure that all data with the same key lands on the same reducer!) A null or blank value is hashed to 0.
Sometimes you can just ignore these records in the join:
select * from foo join bar on foo.bar_id = bar.id and foo.bar_id is not null
This does not always work because sometimes you need to keep the records with the null ids in the query, especially when you are nesting joins. So instead you can do something like this:
select f.*, b.barColumn1, b.barColumn2 from foo f join bar b on f.bar_id = b.id and f.bar_id is not null union all select f.*, '' as barColumn1, '' as barColumn2 from foo f where f.bar_id is null
(Note: It would be tempting to use a left outer join here, but the problem is all the records in "foo" that have a null "bar_id" would still land on the same reducer, even though they do not join to anything.)
Handling blank or null keys is easier because they are easy to filter. What about cases where you legitimately have extreme distributions of data?
I recently encountered an example in a dataset where less than 1% of the records in one table (Table A) joined to 50% of the records in another (Table B). And Table A had millions and millions of records, so I could not explicitly filter out the wayward 1% of records.
The amount of data returned by the top 1% in Table A was so great our webservice could not handle returning it to the client. So we decided to actually prune the top 1% of Table A and only keep the remaining 99% of the records.
My first attempt went something like this:
select
f.*,
barJoin.*
from
foo f
join (
select
b.*,
bc.barCount
from
bar b
join (
select id, count(*) as barCount from bar group by id
) bc on b.id = bc.id
) barJoin on f.bar_id = barJoin.id and barJoin.barCount < 1000
A little complex, but the basic approach was to get a count of how many "bar" records there are per bar "id" value (that is what the inner join is doing). Then when you join foo and bar in the outer join, you can filter based on the count.
But there is a problem. Hive does not allow non equality conditions in joins! It silently ignores them. Yes, you read that correctly. It SILENTLY ignores them. Bummer.
After some experimentation I found two options. The next alternative thing to do is this:
select
f.*,
barJoin.*
from
foo f
join (
select
b.*,
bc.barCount
from
bar b
join (
select id, count(*) as barCount from bar group by id having barCount < 1000
) bc on b.id = bc.id
) barJoin on f.bar_id = barJoin.id
I removed the condition in the join clause and added a having clause to the inner join. This whittles down the invalid "bar" ids before they join, which then also removes the bad keys from "foo" as well.
A good solution if you are all right with throwing away the bad keys from the "foo" table. But what if you want to preserve those "foo" but just omit the joined data from the "bar" table?
select
f.*,
barJoin.*
from
foo f
join (
select
b.*,
bc.barCount,
if (barCount < 1000, 'N', 'Y') as prune
from
bar b
join (
select id, count(*) as barCount from bar group by id
) bc on b.id = bc.id
) barJoin on f.bar_id = barJoin.id and barJoin.prune = 'N'
I removed the having clause, but added an "if" statement to the inner select to create a flag indicating whether or not the joined data should be preserved. The outer join condition can now use the state of the "prune" flag (which is an equality operation!) to select which records should join. In this situation all the keys from foo are preserved, but they only join to bar records if we decided the bar records are small enough.
Please comment with any questions or additional thoughts, email me at smouring@nearinfinity.com, or follow me on Twitter (@marlhammer)!
Published at DZone with permission of Scott Leberknight, DZone MVB. See the original article here.
Opinions expressed by DZone contributors are their own.
Comments