Best unofficial Apache Server developers community
Username
Forgot password?
Sign in with Twitter account
Sign in with Facebook account

COLLECT_SET() in Hive (Hadoop)

⇧
0
⇩
66 views

I just learned about the collect_set() function in Hive, and I started a job on a development 3-node cluster.

I only have about 10 GB to process. The job, however, is literally taking forever. I think there might be either a bug in the implementation of collect_set(), a bug in my code, or the collect_set() function is really resource intensive.

Here's My SQL for Hive (no pun intended):

INSERT OVERWRITE TABLE sequence_result_1
SELECT sess.session_key as session_key,
       sess.remote_address as remote_address,
       sess.hit_count as hit_count,
       COLLECT_SET(evt.event_id) as event_set,
       hit.rsp_timestamp as hit_timestamp,
       sess.site_link as site_link
    FROM site_session sess 
        JOIN (SELECT * FROM site_event 
                WHERE event_id = 274 OR event_id = 284 OR event_id = 55 OR event_id = 151) evt 
            ON (sess.session_key = evt.session_key)
        JOIN site_hit hit ON (sess.session_key = evt.session_key)
GROUP BY sess.session_key, sess.remote_address, sess.hit_count, hit.rsp_timestamp, sess.site_link
ORDER BY hit_timestamp;

There are 4 MR passes. The first took about 30 seconds. The second Map took about 1 minute. And most of the second reduce took about 2 minutes. In the last two hours, it's been incrementing from 97.71% to 97.73%. Is this right? I think there must be some issue. I took a look at the log, and I can't tell whether it's normal.

[Sample of log]

2011-06-21 16:32:22,715 WARN org.apache.hadoop.hive.ql.exec.GroupByOperator: Hash Tbl flush: #hash table = 120894
2011-06-21 16:32:22,758 WARN org.apache.hadoop.hive.ql.exec.GroupByOperator: Hash Table flushed: new size = 108804
2011-06-21 16:32:23,003 INFO org.apache.hadoop.hive.ql.exec.JoinOperator: 4 forwarding 5142000000 rows
2011-06-21 16:32:23,003 INFO org.apache.hadoop.hive.ql.exec.SelectOperator: 5 forwarding 5142000000 rows
2011-06-21 16:32:24,138 INFO org.apache.hadoop.hive.ql.exec.JoinOperator: 4 forwarding 5143000000 rows
2011-06-21 16:32:24,138 INFO org.apache.hadoop.hive.ql.exec.SelectOperator: 5 forwarding 5143000000 rows
2011-06-21 16:32:24,725 WARN org.apache.hadoop.hive.ql.exec.GroupByOperator: Hash Tbl flush: #hash table = 120894
2011-06-21 16:32:24,768 INFO org.apache.hadoop.hive.ql.exec.GroupByOperator: 6 forwarding 42000000 rows
2011-06-21 16:32:24,771 WARN org.apache.hadoop.hive.ql.exec.GroupByOperator: Hash Table flushed: new size = 108804
2011-06-21 16:32:25,338 INFO org.apache.hadoop.hive.ql.exec.JoinOperator: 4 forwarding 5144000000 rows
2011-06-21 16:32:25,338 INFO org.apache.hadoop.hive.ql.exec.SelectOperator: 5 forwarding 5144000000 rows
2011-06-21 16:32:26,467 INFO org.apache.hadoop.hive.ql.exec.JoinOperator: 4 forwarding 5145000000 rows
2011-06-21 16:32:26,468 INFO org.apache.hadoop.hive.ql.exec.SelectOperator: 5 forwarding 5145000000 rows

I'm pretty new at this, and trying to work with collect_set() and Hive Array is driving me off the deep end.

Thanks in advance :)

hadoop mapreduce udf hive
asked June 21, 2011 6:48 pm CDT
Travis Powell
posted via StackOverflow

3 Answers

⇧
0
⇩
 

First thing I would try is getting rid of the sub-select and just join to site_event, then move the event_id filter into the outer where clause and change it to an in(). So something like:

SELECT sess.session_key as session_key,
   sess.remote_address as remote_address,
   sess.hit_count as hit_count,
   COLLECT_SET(evt.event_id) as event_set,
   hit.rsp_timestamp as hit_timestamp,
   sess.site_link as site_link
FROM site_session sess 
    JOIN site_event evt ON (sess.session_key = evt.session_key)
    JOIN site_hit hit ON (sess.session_key = evt.session_key)
WHERE evt.event_id in(274,284,55151)
GROUP BY sess.session_key, sess.remote_address, sess.hit_count, hit.rsp_timestamp, sess.site_link
ORDER BY hit_timestamp;

Additionally, I don't know the sizes of each table, but in general in Hive, you want to keep your largest tables (usually your fact table) on the right hand side of joins to reduce memory usage. The reason being that Hive attempts to hold the left hand side of a join in memory, and streams the right hand side in order to accomplish the join.

answered June 21, 2011 7:27 pm CDT
Jason
⇧
0
⇩
Best answer
 

Major fail. My solution below. There was no issue with the COLLECT_SET after all, it just trying to COLLECT all of the items, of which there were infinite.

Why? Because I joined on something that wasn't even part of the set. Second join used to be the same ON condition, now it correctly says hit.session_key = evt.session_key

INSERT OVERWRITE TABLE sequence_result_1
SELECT sess.session_key as session_key,
       sess.remote_address as remote_address,
       sess.hit_count as hit_count,
       COLLECT_SET(evt.event_id) as event_set,
       hit.rsp_timestamp as hit_timestamp,
       sess.site_link as site_link
    FROM tealeaf_session sess 
        JOIN site_event evt ON (sess.session_key = evt.session_key)
        JOIN site_hit hit   ON (sess.session_key = hit.session_key)
    WHERE evt.event_id IN(274,284,55,151)
GROUP BY sess.session_key, sess.remote_address, sess.hit_count, hit.rsp_timestamp, sess.site_link
ORDER BY hit_timestamp;
answered June 24, 2011 2:10 am CDT
Travis Powell
⇧
0
⇩
 

I'd guess what's happen is that it's producing a COLLECT_SET() for EACH row that would have been returned. So for every row you're returning, it's returning the entire array produced by COLLECT_SET. That might be taxing and taking a long time.

Check the performance with COLLECT_SET out of the query. If that is speedy enough, push the calculation of COLLECT_SET into a sub-query and then use that column instead of doing the calculation where you are.

I haven't used COLLECT_SET or done any tests, from what your post, that's what I'd first suspect.

answered June 24, 2011 2:10 am CDT
Nija

Your answer

Join with account you already have

Sign in with Twitter account
Sign in with Facebook account
Sign in with Google Friend Connect

Preview
Similar questions
Hive, hadoop, and the mechanics behind hive.exec.reducers.max
February 17, 2011
Hadoop/hive metastore
August 7, 2009
MDX support for Hive (Hadoop)
May 23, 2011
Even data distribution on hadoop/hive
July 6, 2010
How should I partition data in s3 for use with hadoop hive?
December 15, 2010
Hadoop Hive Query: Multi-join
June 17, 2011
Establishing a connection between R and a Hive (Hadoop) database
May 19, 2011
Hive / Hadoop / Flatfile: What is an efficient way to combine and concat rows
May 5, 2011
Ability to limit maximum reducers for a hadoop hive mapred job?
February 7, 2011
Using Apache Hive as a MapReduce Input Format and/or Scraping Hive Metadata
April 14, 2011
Hive installation issue while following Hive apache wiki step by step
June 14, 2011
How to import the package org.apache.hadoop.mapreduce.lib.chain in a hadoop 0.20.2 project?
March 25, 2011
Hadoop wordcount unable to run - need help on decoding the hadoop error message
May 17, 2011
A hadoop job complete without map and reduce on a Hadoop Cluster( one namenode ,12 datanode)
June 10, 2011
Cassandra wih Hive
May 9, 2011
Hive with Lucene
January 31, 2011
Linear filter (FIR) in Hadoop (Hadoop in Action exercise)
March 30, 2011
Is hive have its own map reduce program?
May 26, 2011
How to transfer mysql table to hive?
January 7, 2011
Create temporary table in Hive?
March 21, 2011
Hive doesn't respond when I try to make a query
May 18, 2011
Convert Data from HIVE to noSQL
March 6, 2011
What is the advantage of integrating Hbase and Hive
January 6, 2011
Hive - create a table from zip file.
June 16, 2011
Importing data from HDFS to Hive table
April 22, 2011
How to configure hive server to run on remote mode?
June 6, 2011
Can PIG and HIVE be called separate programming models?
February 11, 2011
Running Pig query over data stored in Hive
April 21, 2011
Specifying compression codec for a INSERT OVERWRITE SELECT in Hive
January 28, 2011
What are the consequences of adding a column to an exiting HIVE table ?
February 21, 2011