S-JIS[2012-05-12] 変更履歴

Sqoop注意点

Sqoopを使う上で注意した方が良さそうな点。


分散数

Sqoopでは、MapReduce(Mapタスク)を使って分散してRDBへのアクセスを行う。

普通のMapReduce(TextInputFormatSequenceFileInputFormat)だとファイルのブロック数に応じてタスク数が決まるが、
SqoopではimportでもexportでもMapタスク数を指定する必要がある。(指定しない場合のデフォルトは4)

実際のデータによっては指定したタスク数より小さくなることはあるが、大きくなることは無い。
(importでは投機的実行は行われる)

特にexport(HDFS→RDB)は、入力がHDFS上のファイルになるけれども、ファイルのブロック数などに依存せず、指定されたMapタスク数(小さくなることはある)で実行される。

分散数を増やせば、1タスク当たりの処理件数が減って全体として速くなるはずだ(それがHadoopのコンセプトでもある)が、当然RDBMSに負荷が集中するので、それらの兼ね合いになる。


インデックス

importでは、分散させる為にRDBの特定カラムのデータを使う。
(カラムを特に指定しない場合は対象テーブルのプライマリキーが使われる。プライマリキーが無いテーブルだとエラーになるので、明示的に指定する必要がある)

最初に「select min(カラム), max(カラム) from テーブル」を実行して、対象データの最小値・最大値を取得する。
そして、それを元に各タスクがselectする範囲を決めるので、select条件は「where 値<=カラム AND カラム<=値」の様になるはず。

したがって、データ分割に使用するカラムにはインデックスが構築されていないと、効率が悪い。

exportでもupdate方式(RDB上の既存データを更新するらしい)があったりするので、それを使う場合はインデックスが無いと効率が悪いだろう。


データの偏り

importにおいて、分散させる為に使用するカラムのデータに偏りがある場合、各Mapタスクが対象とする処理件数に偏りが出てしまう。

各Mapタスクが処理する範囲は、指定カラムの最小値と最大値を取得して、Mapタスク数で均等に分けた値になるようだ。
例えば最小値が1、最大値が12、Mapタスク数が4だったとすると、各タスクの処理範囲は1〜3・4〜6・7〜9・10〜12になる。
ここで、実際のデータに4〜6が1件も無いと、4〜6を処理するタスクは0件出力になる。(実質的に他のタスクの負荷が増えて、3タスクで処理しているのと同じになってしまう)
特定のデータ(例えば10)だけデータ件数が突出して多いと、他のMapタスクは早く終わっているのに10〜12のMapタスクだけ遅いという状態になり、その所為でジョブ全体の終了が遅くなる。

したがって、データの偏りがなるべく均等になるようなカラムで分散させた方が良い。

例えば、24時間営業のコンビニで販売時刻をプライマリキーにしている販売明細テーブルがあったとすると、時間帯によってデータ量に差があるだろう。
このとき、販売時刻を分割に使用すると偏りが出てしまう。
この例だと、1ずつ自動的に増加する値のカラムを用意し、それを分割に使用すれば偏りを抑えられそうな気がする。


少量データ

データ量が少ないテーブルからimportする場合でも、MapReduceによってデータが取得される。
そのため、普通のJDBCによるselectなら一瞬で終わるのに、SqoopではHadoop(MapReduceジョブ)の起動で数秒かかることになって、比較すると明らかに効率が悪い。

まぁ、数秒であれば目くじら立てる必要は無いかもしれないが^^;

また、例えば店舗マスターテーブル(店舗コードがプライマリキー、店舗コードの値は1〜12)をimportすることを考えてみる。
店舗コードを分散させるカラムとすると、たかだか12レコードしか無いのに、Mapタスク数4(デフォルト)で実行される。
その結果、HDFS上のファイルはpart-m-00000〜00003の4ファイル(1ファイル当たり3レコード)作られることになる。
すると、それを読み込んで処理するMapReduceプログラムは、4タスクに分割して処理を行うことになる。MapReduce的には効率が悪いと思われる。

まぁ、こういう少量データのファイルのみを読み込んで処理するMapReduceというのは考えにくいか^^;


投機的実行

Hadoop(MapReduce)には投機的実行という機能がある。
処理が遅いタスクがあったら、全く同じタスクを別のノードで実行し、早く終わった方の結果を採用するというもの。

Sqoopのimportでも投機的実行は行われる。
selectなので取得結果やRDB上のデータへの影響は無いものの、RDBMSへの負荷は余計にかかることになる。
RDBMS側が投機的実行の分の負荷に耐えられるなら問題ないと思うが…。

さすがにexportでは投機的実行は実施されないようになっている。
(基本はSQLのinsertなので、重複実行されたらユニークインデックスの重複エラーになるか、データが多重登録されてしまう^^;)

投機的実行をするかどうかは、Hadoopのconfのmapred.map.tasks.speculative.executionおよびmapred.reduce.tasks.speculative.executionで設定できるようだ。
(SqoopではMapタスクしか実行されないので、前者のプロパティーだけでいいはずだが、後者のreduceの設定もされている)
importではtrue、exportではfalseになっている。


ダイレクトモード

普通にSqoopを使う場合は汎用SQL(selectやinsert・update)で処理されるが、「--direct」を指定して、RDBMS固有の機能を使って処理させることが出来る。

ただし、ダイレクトモードはRDBMSの種類によって出来る事・出来ない事があるので、RDBMS毎に個別に見る必要がある。
(例えば、MySQLもPostgreSQLもダイレクトインポートはSequenceFileに対応していないが、-as-sequencefileを指定すると、MySQLではエラーになるがPostgreSQLでは無視してテキストファイルが作られる。それくらい統一しろよって感じだが^^;


Sqoop目次へ戻る / Hadoopへ行く / 技術メモへ戻る
メールの送信先:ひしだま