E-MapReduce:Hive job optimization

Cleanse data

Reduce the volume of data that Hive processes before expensive operations run.

1. Filter by partition when reading from a partitioned table. This avoids full table scans.

2. Filter data before a JOIN operation, not after.

3. Store repeatedly used temporary results in an intermediate table to avoid redundant computation.

Replace multiple DISTINCT operators

Multiple DISTINCT operators in a single query can cause data expansion. Replace them with two GROUP BY clauses: an inner GROUP BY to deduplicate and an outer GROUP BY to aggregate.

Before optimization

SELECT k,
       COUNT(DISTINCT CASE WHEN a > 1 THEN user_id END) user1,
       COUNT(DISTINCT CASE WHEN a > 2 THEN user_id END) user2,
       COUNT(DISTINCT CASE WHEN a > 3 THEN user_id END) user3,
       COUNT(DISTINCT CASE WHEN a > 4 THEN user_id END) user4
FROM t
GROUP BY k;

After optimization

The inner query groups by (k, user_id) to deduplicate, reducing the data volume. The outer query groups by k to produce the final counts.

SELECT k,
       SUM(CASE WHEN user1 > 0 THEN 1 ELSE 0 END) AS user1,
       SUM(CASE WHEN user2 > 0 THEN 1 ELSE 0 END) AS user2,
       SUM(CASE WHEN user3 > 0 THEN 1 ELSE 0 END) AS user3,
       SUM(CASE WHEN user4 > 0 THEN 1 ELSE 0 END) AS user4
FROM
        (SELECT k,
                user_id,
                COUNT(CASE WHEN a > 1 THEN user_id END) user1,
                COUNT(CASE WHEN a > 2 THEN user_id END) user2,
                COUNT(CASE WHEN a > 3 THEN user_id END) user3,
                COUNT(CASE WHEN a > 4 THEN user_id END) user4
        FROM t
        GROUP BY k, user_id
        ) tmp
GROUP BY k;

Handle data skew in GROUP BY operations

Use one of the following methods to handle skewed keys.

SELECT * FROM log a LEFT OUTER
JOIN bmw_users b ON
CASE WHEN a.user_id IS NULL THEN CONCAT('dp_hive',RAND()) ELSE a.user_id=b.user_id END;

Map stage memory

Reduce stage memory

Map tasks

In Hadoop Distributed File System (HDFS), each file is stored as data blocks. The number of data blocks is one of the factors that determine how many map tasks run. In most cases, each map task reads one data block.

• Too many small files: Reduce the map task count to improve resource utilization.

• Few large files: Increase the map task count to reduce per-task workload.

The key parameters are mapred.map.tasks, mapred.min.split.size, and dfs.block.size.

Reduce tasks

Two approaches control the reduce task count:

• Set bytes per reducer. Use hive.exec.reducers.bytes.per.reducer to let Hive calculate the count automatically:

    reducer_num = min(total_size / hive.exec.reducers.bytes.per.reducer, hive.exec.reducers.max)

• Set the count directly. Use mapred.reduce.tasks to specify an explicit number of reduce tasks.

Starting and initializing reduce tasks consumes time and resources. Each reduce task generates one output file. An excessive number of reduce tasks leads to many small files, which can cascade when those files become input for downstream tasks.