Hive Auto Convert Join

これか！？

CDHのバージョンを4.7から5.3.2に上げたときに，Hiveも0.13.1に上がり，オプションのデフォルト値が変わっていたのを見落としていた。。

joinの最適化に関連するオプションのようで，以下の値がデフォルトでtrueに変更されていた。どうも，テーブル結合するクエリで落ちることがあると思ったらこれが関連していたっぽい。

<property>
<name>hive.auto.convert.join</name>
<value>true</value>
<description>Whether Hive enables the optimization about converting common join into mapjoin based on the input file size</description>
</property>

<name>hive.auto.convert.join</name>

<description>Whether Hive enables the optimization about converting common join into mapjoin based on the input file size</description>

</property>

https://cwiki.apache.org/confluence/display/Hive/LanguageManual+JoinOptimization

上記ドキュメントにも書かれているように，このオプションがtrueの時，実行時にjoinの最適化が行われる。他にも関連しそうなオプションがいくつかあり，オプションが何を示しているのかよく分からなかったので，ソースも少し追ってみる。（が理解は十分ではない。。）

このオプションが参照されている箇所を探すと，org.apache.hadoop.hive.ql.optimizer.physical.PhysicalOptimizerのinitializeメソッドが見つかる。メソッドの中で，オプションがtrueの時，CommonJoinResolverがresolverリストに加えられている。そして，このresolverの先でTaskTreeを走査し，MapJoinタスクへの変換が試みられている。

if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVECONVERTJOIN)) {
resolvers.add(new CommonJoinResolver());
// The joins have been automatically converted to map-joins.
// However, if the joins were converted to sort-merge joins automatically,
// they should also be tried as map-joins.
if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_AUTO_SORTMERGE_JOIN_TOMAPJOIN)) {
resolvers.add(new SortMergeJoinResolver());
}
}

if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVECONVERTJOIN)) {

resolvers.add(new CommonJoinResolver());

// The joins have been automatically converted to map-joins.

// However, if the joins were converted to sort-merge joins automatically,

// they should also be tried as map-joins.

if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_AUTO_SORTMERGE_JOIN_TOMAPJOIN)) {

resolvers.add(new SortMergeJoinResolver());

}

タスクTreeのnodeは，CommonJoinTaskDispatcherにdispatch（processCurrentTaskメソッド）されている。そのprocessCurrentTaskメソッドの中でMapJoinタスクへの変換可否がチェックされ，可能ならば変換を行なう。その際にざっと以下の流れで処理が行われているようだった。

テーブルのinput pathsを調べ，既知のデータサイズの合計値を計算する
bigTableCandidatesの取得
hive.auto.convert.join.noconditionaltaskがtrueの場合，bigTableCandidatesリストからbigTableを求める。
bigTableCandidatesのリストから１件ずつ自身以外の既知のテーブルサイズの合計を求め，hive.auto.convert.join.noconditionaltask.sizeで指定されたしきい値を超えていないかチェックする。
しきい値を超えていなければ，最大サイズのテーブルかチェックする。
bigTableが見つかれば，MapJoinTaskへ変換する。（ConditionalTaskは生成されない）
3で候補が見つからない場合，smalltablesの合計サイズがhive.mapjoin.smalltable.filesizeのしきい値を超えていないかチェックし，しきい値を超えていない場合，MapJoinTaskへ変換される。（ConditionalTaskが生成される）

Hashテーブルのサイズは，hiveクライアントを実行したプロセスのメモリ？が使われる。環境変数でなく，hive.mapred.local.memというプロパティを指定するとそのサイズを変えられそうなのだが，cloudera manager環境下なのだがGatewayのサイズが優先されているように見える。HADOOP_CLIENT_OPTSを使えば適用できそうだった。

HADOOP_CLIENT_OPTS="-Xmx1000m" hive -e "select ...."

1	HADOOP_CLIENT_OPTS="-Xmx1000m" hive -e "select ...."

ソースももう少しじっくり読み，中身の理解を深めたい。。

参考

コメント