Hive on TezにおいてDISTRIBUTE BY指定したクエリのFile Mergeの挙動

Hive Java

Hive on Tezにおいて、DISTRIBUTE BYを指定したクエリが、File Mergeが有効化されていたことによって想定外の挙動になった。 Hiveのバージョンは、3.1.3でORC形式のテーブルを想定している。また、他のバージョンでは修正されている可能性はある。

まず、DISTRIBUTE BYの挙動からおさらいしておくと、DISTRIBUTE BYに指定したカラムについて、カラムの値が同じであれば同じReducerで処理されることを保証するものである。また、競合を避けるためにReducerの単位で別のファイルに結果が書かれることになる。 つまりは、DISTRIBUTE BYで指定したカラムについて、同じ値を持つレコードは、必ず同じファイルに書かれるということが保証される。

これはHive on TezでもHive on MapReduceでも同じことである。

HiveではORDER BYをつけることでデータ全てが一貫してソートされているグローバルソートを実現できるが、全てのデータを1箇所でソートする必要があるため、当然Reducerの数も1つに制限されてしまう。書き込み先のファイルが1つになるため、データサイズが大きい場合に容易にボトルネックとなる。 そこで、グローバルソートは不要だが、特定のカラム単位では一貫したソート結果が欲しいと言うようなケースで、このDISTRIBUTE BYとSORT BYを組み合わせることで、Reducerごとのソートを可能とすることができる。

しかし、Small FIle Problemを避けるためにhive.merge.tezfilesがtrueになっている環境下においては、この挙動が変わってしまうと言うことが観測された。 hive.merge.tezfilesをtrueにしていると、そうではない場合のFile MergeのためのReducerが追加される。

ここまでは良いが、Hive3.1.3では、その前段までは保持されていたDISTRIBUTE BYのコンテキストが、File Mergeの時点では失われてしまい、完全に無視されてファイルマージが行われてしまうということが分かった。 つまり、ファイルマージされる前まではReducer単位でのソートができていたが、その後ファイルサイズを最適化するために、前段のファイルを分割/結合して、マージすることになる。 ファイル単位での単なる結合であれば、これはReducer毎のソートを維持したまま達成可能である。しかし分割が行われた時点でこの保証は無くなってしまう。 ORCのストライプレベルでのファイルマージを制御するhive.merge.orcfile.stripe.level フラグをfalseにした場合でも、やはりファイル分割が行われてしまった。

結果として、Reducer毎のソートは無効になってしまい、異なるファイルに同じDISTRIBUTE BYカラムを持つレコードが分割して含まれてしまうと言う問題が発生する。

こういった問題を避けつつ、Small File Problemも一定数解決したい場合には、

set hive.merge.tezfiles=false;
set hive.exec.reducers.max=xxx;

のように指定を行い、ファイルマージを回避しつつ、Reducerの数を減らすことでなるべくパフォーマンスを落とさずにファイル数を少なくすると言うチューニングが考えられる。