Hive join query optimisation

Question

Table A
---------
col1, col2,Adate,qty

Table B
-------
col2,cost,Bdate

The table sizes are as follows:

A: 1 million B: 700k

Consider this query:

SELECT 
  A.col1,
  A.col2,
  B.Bdate bdate,
  SUM(qty)*COLLECT_LIST(cost)[0] price 
FROM A 
JOIN B 
ON (A.col2 = B.col2 AND A.Adate <= B.Bdate) 
GROUP BY 
  A.col1,
  A.col2,
  B.bdate;

The above hive query takes more than 3 hrs on a cluster of 4 slaves(8GB memory,100 GB disk) and 1 master(16 GB memory, 100 GB disk)

Can this query be optimized? If yes, where can the optimization be possible?

Answer 1

Use Tez and mapjoin.

set hive.auto.convert.join=true; --this enables map-join
set hive.mapjoin.smalltable.filesize=25000000; --adjust for your smaller table to fit in memory
set hive.execution.engine=tez;

Also this computation is not memory-efficient:

SUM(qty)*COLLECT_LIST(cost)[0] price 

COLLECT_LIST will collect all cost values in the group into non unique(contains values from ALL rows in the group) and unordered (yes, unordered, because you have no any distribute + sort before collect_list) array. This array can be big enough (the number of elements = the number of rows in the group), depending on your data, then you are taking [0] element, it means that you are picking just any random cost from the group. Does it make any sense to collect array to get just any random element? Use min() or max instead. If it does not matter which cost should be taken, then min(cost) or max(cost) or some other scalar function will consume less memory. You can use first_value analytic function (may require sub-query, but it will be memory-efficient also)

Answer 2

I will try to give you some advices to improve query performance in Hive.

• Check the execution engine you are using

set hive.execution.engine;

If you execution engine is mr, rather than MapReduce, you may be able to use Apache Spark or Apache Tez, both of which are faster than MapReduce.

set hive.execution.engine=tez;

• Join queries are computationally expensive and can be slow, especially when you’re joining three or more tables, or if you’re working with very large data.

One strategy that can be used to remedy this problem is to join the data in advance and store the pre-joined result in a separate table, which you can then query. this is one way of denormalizing a normalized database, to make it easier to run analytic queries. This approach of pre-joining tables has some costs, but it can make analytic queries easier to write and faster to run.

There are some other techniques for improving Hive query performance

• Join table ordering (Largest table last)

As with any type of tuning, it is important to understand the internal working of a system. When Hive executes a join, it needs to select which table is streamed and which table is cached. Hive takes the last table in the JOIN statement for streaming, so we need to ensure that this streaming table is largest among the two.

A: 1 million B: 700k

Hence, when these two tables are joined it is important that the larger table comes last in the query.

• Bucketing stores data in separate files, not separate subdirectories like partitioning.

It divides the data in an effectively random way, not in a predictable way like partitioning. When records are inserted into a bucketed table, Hive computes hash codes of the values in the specified bucketing column and uses these hash codes to divide the records into buckets. For this reason, bucketing is sometimes called hash partitioning. The goal of bucketing is to distribute records evenly across a predefined number of buckets. Bucketing can improve the performance of joins if all the joined tables are bucketed on the join key column.

For more on bucketing, see the page of the Hive Language Manual describing bucketed tables,

BucketedTables

bucketing-in-hive

Partitioning

• Partitioning is a way of dividing a table into related parts based on the values of particular columns like date, city, and department.

Each table in the hive can have one or more partition keys to identify a particular partition. Using partition it is easy to do queries on slices of the data.

apache-hive-partitions