たごもりすメモ

コードとかその他の話とか。

Hive dynamic partition insertsにまつわるいくつかの問題と対処について

だいぶ前のHiveの機能準拠で作ってたクラスタを大幅に作り直したので、ついでにETL処理をdynamic partition inserts一発でやればMapReduce 1ジョブで済んで超効率的に！ やった！ と思ったらいくつかハマったのでメモ。

なおdynamic partition insertsについては説明が面倒なので公式Wikiの該当ページを読むとよい、が、簡単に言うとHiveでパーティションにINSERTするときにINSERT先のパーティション指定をSELECTクエリの出力により行う、というもの。

なお断りがない限りは HDP2.1 with Hive 0.13 の環境で試したものとする。(移行元はCDH4)

クエリの書き方

単純に言うと、パーティションとして指定したいカラムは SELECT 句の最後に置かなければならない。

簡単に言うと year=INT/month=INT/day=INT というパーティション階層になっているテーブルにINSERTするとき、month と day をSELECT句から取り出したい場合には、以下のようにクエリを書かなければならない。

INSERT OVERWRITE TABLE tbl PARTITION (year=2014, month, day)
SELECT
  data1, data2, data3, data4,
  month, day
FROM source_table WHERE ...

dynamicに指定したいカラムだけを書いておく。

ハマった

いちばんわかりにくくハマったのは INSERT クエリを流したとき、なぜか毎回いくつかのReducerだけが他のReducerよりもはるかに長く動き続けており、全体の処理の規模が大きいときはそれこそ永遠にお待ちください状態になる、という挙動。

最初は MRv2 on YARN 移行に伴うものかと思ったが、違うっぽい。よく見てみると毎回 shuffle で莫大な時間を消費しているようだった。*1設定をあれこれ見ていたらこんなのが見付かった。

hive.optimize.sort.dynamic.partition

  • Default Value: true in Hive 0.13.0 and 0.13.1; false in Hive 0.14.0 and later (HIVE-8151)
  • Added In: Hive 0.13.0 with HIVE-6455

When enabled, dynamic partitioning column will be globally sorted. This way we can keep only one record writer open for each partition value in the reducer thereby reducing the memory pressure on reducers.

Configuration Properties - Apache Hive - Apache Software Foundation

実は最初にCDH4のクラスタでほぼ同じことをやったときは問題なく終わってて、HDP 2.1 のクラスタでやったときにはまった。このせいっぽい。Hive 0.13 においてはこのオプションがデフォルトで有効になっており、dynamic partition insertsを使うときに自動的に出力をsortにかける。しかもその record writer は必ずひとつのreducerとなる。
このため、出力先のパーティションごとに考えたときにこのサイズが巨大なものがあると、sortがいつまでも終わらず非常に長い時間がかかる、ということらしい。

気付いてこれを false にしたらパーティション全体に対するsortが行われなくなり、reducerが走らずそのまま出力がパーティションに書かれるようになったので、妙に時間がかかるようなこともなく終わるようになった。めでたしめでたし。

なお Hive 0.14 ではデフォルト false になっている。やっぱみんなハマったんじゃないかな。

他にもいくつか

Hive on MRv2 on YARNで気になったところがあったので書いておく。

hive.exec.dynamic.partition
  • Default Value: false
  • Added In: Hive 0.6.0
  • Whether or not to allow dynamic partitions in DML/DDL.

有効にしましょう。

hive.exec.dynamic.partition.mode
  • Default Value: strict
  • Added In: Hive 0.6.0
  • In strict mode, the user must specify at least one static partition in case the user accidentally overwrites all partitions. In nonstrict mode all partitions are allowed to be dynamic.

簡単に言うと、dynamic partition inserts時、strict になっていれば、少なくとも1段のパーティションについてはクエリ内で明示的に指定されていなければならない(SELECT結果を使わずに指定しなければならない)。

これのパーティション指定用カラムをトップレベルからすべて動的に指定したい場合、このモードを nonstrict として指定してやる必要がある。

SET hive.exec.dynamic.partition.mode=nonstrict;

INSERT OVERWRITE TABLE tbl PARTITION (year, month, day)
SELECT
  data1, data2, data3, data4,
  year, month, day
FROM source_table WHERE ...

トップレベルのパーティションを無尽蔵に増やしてしまうと管理がものすごく大変になるから、親切心でデフォルトを strict にしてあるんだと思う。DROP PARTITIONも面倒になるし。覚悟がある人だけ nonstrict にしましょう。

hive.exec.max.dynamic.partitions
  • Default Value: 1000
  • Added In: Hive 0.6.0
  • Maximum number of dynamic partitions allowed to be created in total.

1回のdynamic partition insertsクエリで生成されるパーティション数の上限。日付とかでpartitionを作っている分には超えないが、何かしらの商品コード等を用いてパーティション作成する場合にはけっこうひっかかるかもしれない。注意。
ひっかかったらクエリ失敗時に例外にメッセージ出てるはずだから、増やそう。

hive.exec.max.dynamic.partitions.pernode
  • Default Value: 100
  • Added In: Hive 0.6.0
  • Maximum number of dynamic partitions allowed to be created in each mapper/reducer node.

1回のdynamic partition insertsクエリにおいて、ひとつのmapper/reducerで作られるパーティション数の上限。前項と同じでひっかかるケースはあると思う。注意。ひっかかったら上げる、でいいと思う。

hive.exec.reducers.max と mapreduce.job.reducer

hive.exec.reducers.max はデフォルトが999で、この数字は mapreduce.job.reducer よりも優先される。これがわかってなかったので、なぜか毎回 reducers が999になり、あれー？ と思っていた。注意。 oza先生に教えておもらいました。多謝。

が、これは前述の hive.optimize.sort.dynamic.partition=false により意味がなくなった。

結論

みんなで hive.execution.engine=tez を指定しよう！

*1:これMRv1 JobTrackerの画面に較べてMRv2の画面に出てこなくなったので、だいぶ長いこと気付かなかった