HBase データソースに対する Pushdown (1)

この記事は Apache Drill Advent Calendar 2015 の15日目の記事です。

HBase は Hadoop 上で動作する、「ワイドカラム型」NoSQL データベースです。RDBMS 風のテーブル構造を持ちますが、固定のスキーマを持つわけではないのでデータ構造の変更には柔軟である一方、HBase 自体にはシンプルなアクセス API があるのみで SQL クエリ機能はありません。Drill は HBase をデータストアとして使用することができるため、HBase に格納されたデータを使い慣れた SQL で柔軟に取り扱えるようになるのがメリットです。

ところが、Drill のようにデータストアと実行エンジンが分離しているアーキテクチャで、最適化をまったく考えずに実装してしまうと性能が全然出ません。最も単純な実装として、HBase でテーブルをフルスキャンして Drill 側にデータを渡し、Drill が SQL クエリの実行に必要なフィルタリングや集計を行う、という役割の分離が考えられますが、HBase のフルスキャンかなり遅いです。まあ、これは HBase に限らず Drill が使用するどのようなデータストア（MongoDB や RDBMS とか）にも当てはまります。

そこで、このような状況では一般的に Pushdown という最適化が行われます。HBase の例で言うと、まず HBase では「テーブルの各行にはユニークな行キーが存在しており、ソートされインデックスがついた状態で格納されている」という前提があるため、クエリの述語（WHERE 句の条件）によってあらかじめ行キーのスキャン範囲を絞り込めれば、I/O アクセスが大幅に減り性能向上につながります。他にも、カラムの値の比較が必要な場合に、HBase 内部の実装で比較を行ってから結果を Drill 側に渡してやったほうがより効率的で高速です。このように、より低いレイヤーの実装に処理を移譲する最適化が Pushdown です。

最近 BigQueryで150万円溶かした人の顔というエントリで、「WHERE句には何を書いてもテーブルをフルスキャンしてしまう」というのが話題になっておりましたが、BigQuery に述語の Pushdown の実装がなされていればこんなことにはならなかったはずです。Drill は BigQuery のベースになっている Dremel のオープンソース実装なので、Drill にできて BigQuery にできないことはないと思うのですが、謎ですね。

さて、実際にどのようなクエリがどんな風に Pushdown されるのかを見てみましょう。EXPLAIN PLAN FOR を SELECT の前につけることでクエリの実行プランが表示されるので、違いを確認することができます。

まず HBase テーブル全体を SELECT するクエリ。

$ apache-drill-1.3.0/bin/drill-embedded
0: jdbc:drill:zk=local> EXPLAIN PLAN FOR
. . . . . . . . . . . > SELECT row_key, cf.c1
. . . . . . . . . . . > FROM hbase.`table1`;
+------+------+
| text | json |
+------+------+
| 00-00    Screen
00-01      Project(row_key=[$0], EXPR$1=[$1])
00-02        Project(row_key=[$0], ITEM=[ITEM($1, 'c1')])
00-03          Scan(groupscan=[HBaseGroupScan [HBaseScanSpec=HBaseScanSpec [tableName=table1, startRow=null, stopRow=null, filter=null], columns=[`row_key`, `cf`.`c1`]]])
...

そして WHERE 句で行キーに条件をつけた場合のクエリ。

0: jdbc:drill:zk=local> EXPLAIN PLAN FOR
. . . . . . . . . . . > SELECT row_key, cf.c1
. . . . . . . . . . . > FROM hbase.`table1`
. . . . . . . . . . . > WHERE row_key = 'xxx';
+------+------+
| text | json |
+------+------+
| 00-00    Screen
00-01      Project(row_key=[$0], EXPR$1=[$1])
00-02        Project(row_key=[$0], ITEM=[ITEM($1, 'c1')])
00-03          Scan(groupscan=[HBaseGroupScan [HBaseScanSpec=HBaseScanSpec [tableName=table1, startRow=xxx, stopRow=xxx\x00, filter=null], columns=[`row_key`, `cf`.`c1`]]])
...

実行プランの Scan オペレーターのパラメーターの中に HBaseScanSpec という指定があり、さらに startRow, stopRow という指定もついていることがわかります。そして、WHERE 句をつけると startRow, stopRow に値が入ることも確認できます。

実際には、HBase の API で Scan オブジェクトが作られて、スキャンの開始行キー startRow、終了行キー stopRow がパラメータとして渡され、HBase の範囲スキャン操作が行われます。WHERE 句に行キーの条件をつけることで、大幅に処理が効率化されることがわかります。

別の例も見てみましょう。

0: jdbc:drill:zk=local> SELECT row_key, cf.c1
. . . . . . . . . . . > FROM hbase.`table1`
. . . . . . . . . . . > WHERE cf.c1 = 'xxx';
+------+------+
| text | json |
+------+------+
| 00-00    Screen
00-01      Project(row_key=[$0], EXPR$1=[$1])
00-02        Project(row_key=[$0], ITEM=[ITEM($1, 'c1')])
00-03          Scan(groupscan=[HBaseGroupScan [HBaseScanSpec=HBaseScanSpec [tableName=table1, startRow=null, stopRow=null, filter=SingleColumnValueFilter (cf, c1, EQUAL, xxx)], columns=[`row_key`, `cf`.`c1`]]])
...

今度は行キーの条件はつけていないため範囲スキャンは行われませんが、カラムの値の比較条件がついていることで、filter というパラメータに SingleColumnValueFilter の指定が出てきます。これは HBase のスキャン時に、そのまま SingleColumnValueFilter のパラメータとして HBase 内部のフィルタリング処理に使われます。

以上のように、Pushdown によるクエリ実行計画の最適化の様子がおわかりいただけたかと思いますが、クエリの書き方によっては Pushdown が期待通りに働かないケースもあります。そのあたりの注意点は次の記事に続きます。