Optimize a long running hive query - has a join with same table

[#1]

The below query takes a lot of time to execute. It is run with tez execution engine.

SELECT STG.EMP_TYPE,DEPT,COUNT(DISTINCT EMP_ID) AS COUNT, A.TOTAL_COUNT
FROM STAGE_SOURCE STG 
LEFT OUTER JOIN 
(SELECT EMP_TYPE,COUNT(DISTINCT EMP_ID) AS TOTAL_COUNT FROM STAGE_SOURCE GROUP BY EMP_TYPE) A
ON STG.EMP_TYPE = A.EMP_TYPE
GROUP BY STG.EMP_TYPE,DEPT,A.TOTAL_COUNT;

Is there any rewrite option or optimization strategy which can improve the query performance?

The subquery with alias "A" itself takes 2-3hrs to execute.

Attaching the explain plan of just the join subquery "A"

explain-plan.txt.

[#2]

@Sooraj Antony Please attach explain plan. You can copy the plan in text and upload in pdf format.

[#3]

The COUNT(DISTINCT) could be the bottleneck if it is not being parallelized. Can you share the explain plan ?

[#4]

@Neeraj Sabharwal @Jean-Philippe Player Explain Plan file is attached..

[#5]

some obvious optimizations would be to convert the tables to ORC, you're using text, the tables are pretty large and ORC can help by skipping batches of rows that don't match the criteria. Another thing to look at is table ordering, here's more info https://cwiki.apache.org/confluence/display/Hive/OuterJoinBehavior and https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=34015666

[#6]

Table

STAGE_SOURCE

is already in ORC format.

[#7]

please check, if set hive.map.aggr=true or not.

Please change the table to ORC format and try to run only subquery. I hope, results should come fast.

Please check, if table can be bucketed on emp_type or not. try to implement SMB join.

[#8]

The table is in ORC and haven't tried SMB, but the process is getting stuck at the last reducer and the whole processing runs in a single node while other nodes are idle.

Can you explain the function of "hive.map.aggr" parameter?

[#9 · Accepted solution]

@Sooraj Antony

I believe the problem is most of records has same value for EMP_TYPE (aka skewed records), that would cause all records from same EMP_TYPE value to be sent to same reducer and would cause the last reducer to take a long time to finish.

Assuming you have small number of different values of EMP_TYPE (and the result fit in memory), try solution below:

set hive.ignore.mapjoin.hint=false;

SELECT /*+ MAPJOIN(A) */ STG.EMP_TYPE,DEPT,COUNT(DISTINCT EMP_ID) AS COUNT, A.TOTAL_COUNT
FROM STAGE_SOURCE STG 
LEFT OUTER JOIN 
(SELECT EMP_TYPE,COUNT(DISTINCT EMP_ID) AS TOTAL_COUNT FROM STAGE_SOURCE GROUP BY EMP_TYPE) A
ON STG.EMP_TYPE = A.EMP_TYPE
GROUP BY STG.EMP_TYPE,DEPT,A.TOTAL_COUNT;

[#10]

@Sooraj Antony has this been resolved? Can you post your solution or accept the best answer?