E-MapReduce:Hive ジョブの最適化

データのクレンジング

Hive がコストのかかる操作を実行する前に、処理するデータ量を削減します。

1. パーティションテーブルから読み込む際に、パーティションでフィルタリングします。これにより、全表スキャンを回避できます。

2. JOIN 操作の前にデータをフィルタリングし、後にはフィルタリングしません。

3. 繰り返し使用される一時的な中間結果は中間テーブルに保存し、冗長な計算を回避します。

複数 DISTINCT オペレーターの置き換え

単一クエリ内の複数 DISTINCT オペレーターは、データ拡張を引き起こす可能性があります。これらを2つの GROUP BY 句に置き換えます。内部 GROUP BY で重複排除し、外部 GROUP BY で集約します。

最適化前

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;

最適化後

内部クエリは (k, user_id) でグループ化して重複排除し、データ量を削減します。外部クエリは k でグループ化して最終的なカウントを生成します。

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;

GROUP BY 操作におけるデータスキューの処理

スキューしたキーを処理するには、以下のいずれかの方法を使用します。

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;

マップステージメモリ

リデュースステージメモリ

マップタスク

Hadoop 分散ファイルシステム (HDFS) では、各ファイルはデータブロックとして保存されます。データブロック数は、実行されるマップタスク数を決定する要因の1つです。ほとんどの場合、各マップタスクは1つのデータブロックを読み込みます。

• 小さなファイルが多すぎる場合: マップタスク数を削減して、リソース利用率を向上させます。

• 大きなファイルが少ない場合: マップタスク数を増加させて、タスクごとのワークロードを削減します。

主要パラメーターは mapred.map.tasks、mapred.min.split.size、および dfs.block.size です。

リデュースタスク

リデュースタスク数を制御する2つのアプローチがあります。

• リデューサーあたりのバイト数を設定。 hive.exec.reducers.bytes.per.reducer を使用して、Hive にカウントを自動的に計算させます。

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

• カウントを直接設定。 mapred.reduce.tasks を使用して、明示的なリデュースタスク数を指定します。

リデュースタスクの開始と初期化には時間とリソースを消費します。各リデュースタスクは1つの出力ファイルを生成します。過剰なリデュースタスク数は多くの小さいファイルにつながり、これらのファイルがダウンストリームタスクの入力になるとカスケードする可能性があります。