前回、Apache Drillをビルドして動かし(http://qiita.com/qtwi/items/9f404a4434e153419629) 、HDFS上のファイル、Hiveのデータ、mongodbのデータを統一的にSQLクエリできるようになった。
高速かつ、便利でよいのだが、まだサポートされるデータベースが少なくて、何でもDrillで、とはいかない。いずれはDrillのストレージプラグインが開発されてくるだろうが、現存すぐにあるデータ、例えば、広く使われている MySQLなどのデータをなんとかしたいと思うだろう。

Apache Sqoopを動かして、MySQLのデータをHDFSやHiveに取り込んでしまえば、一手間かかるが、Drillを使って、MySQLのデータも扱えるようになる。他にも SQL ServerやOracleのデータを取り込める。取り込む先は、HDFSやHive 以外に HBase、accumulo へと取り込める。

今回、ビルドする Sqoop はバージョン 1.4.5 で、開発が進む 1.99ではないが、こなれていて、簡単にビルドして動かすことができる。
オリジナルの Sqoop 1.4.5 では、Hadoop 2.0（とそれ以前）と Hive 0.13.0 に対応するが、ここでは、この記事シリーズで使っている、Hadoop 2.6.0(http://qiita.com/qtwi/items/a6d282ad49685ce62b64) とHive 0.12.0(http://qiita.com/qtwi/items/59e135b0a4c46e37c876) に対応させてビルドする。
今回も、MSYS/MinGW環境での動作となるが、Sqoop は比較的単純で、MSYS対応の特別な作業はしていない。

ビルドが面倒な場合、以下から、Drill、HiveやHadoopなども含め、ビルド済みバイナリを入手できる。

(http://sourceforge.jp/projects/win-hadoop/releases/62852)

Sqoop のビルド

Sqoop のソースを Apache Archives からダウンロード(https://archive.apache.org/dist/sqoop/1.4.5/sqoop-1.4.5.tar.gz) して展開する。
Githubでは、アーキテクチャの異なる1.99以降の開発が進んでいるので、アーカイブをダウンロードしたほうが簡単だろう。
パッチ(https://raw.githubusercontent.com/qtwi/msys-sqoop-1.4.5/master/patch/sqoop-1.4.5-2.6.0-0.12.0-src-0001.patch) を入手して適用する。

mv sqoop-1.4.5 sqoop-1.4.5-src
cd sqoop-1.4.5-src
patch -u -p1 < ../sqoop-1.4.5-2.6.0-0.12.0-src-0001.patch

今回は、デフォルトのHadoop バージョン、Hive バージョンをそれぞれ 2.6.0、0.12.0 として設定済みなので、これらのオプションなくビルドできる（テストの実行は失敗するので、予めビルドプロセスから外してある）。

ant clean package

動作に問題はないようだが、デフォルトの zookeeper のバージョンが若干古く、これを 3.4.6 に指定したり、あるいは、hadoopのバージョンを指定してビルドするなら、以下のようにする。

ant package -Dhadoopversion=260 -Dhadoop.version.full=2.6.0 -Dzookeeper.version=3.4.6

Hive のバージョン（sqoop の扱い上、正確には、hcatalog のバージョン）は、0.12.0 に合わせて修正してしまっているので、0.13.0 を指定してビルドすることはできない。

ビルドが終了すると、build/sqoop-1.4.5.bin__hadoop-2.6.0 のようなフォルダに sqoop 一式がビルドできているので、このフォルダを任意のアプリのフォルダにコピーする。

mv build/sqoop-1.4.5.bin__hadoop-2.6.0 c:/apache

Sqoop の設定

MySQL ドライバーのダウンロード

Sqoop では、MySQL などのデータベースとの接続に JDBCドライバーが必要となるが、これらがビルド結果に入っていないので、別途ダウンロードして、lib フォルダにコピーする必要がある。

curl -LO http://central.maven.org/maven2/mysql/mysql-connector-java/5.1.34/mysql-connector-java-5.1.34.jar
mv mysql-connector-java-5.1.34.jar c:/apache/sqoop-1.4.5.bin__hadoop-2.6.0/lib

conf/sqoop-site.xml

いくつかプロパティを指定できるが、今回は特に指定していない。実数の扱い方を変更するオプションなどがあるが、デフォルトで問題ないと思う。
いくつかある JVMプロパティは、他のJavaアプリ同様、コマンドラインから -D オプションで指定できる。

hive.cmd の作成

後で、Hive に取り込む作業をするが、そのとき、Windows では、Sqoop 内部から、 $HIVE_HOME/bin/hive.cmd が呼び出される。しかし、Hive 0.12.0 には、hive.cmd がないのでエラーとなってしまう。
そこで、hive.cmd を以下のような内容で作成しておく。

$HIVE_HOME/bin/hive.cmd

setlocal
set PATH=C:\MinGW\MSYS\1.0\bin;%PATH%
set HIVE_HOME=C:/apache/hive-0.12.0
bash %HIVE_HOME%/bin/hive %*

MSYS/MinGW のパスを追加しているが、すでにシステムのパスに追加してあれば必要ない。HIVE_HOME もシステムグローバルにあれば必要はないが、bash の引数として指定する都合上、パス区切りは、スラッシュ '/' である必要がある。

オプションファイルの設定

利用しなくてもよいが、取り込むデータベースや取り込み方に合わせて、オプションファイルを書いておくことができる。

mysql-import.txt

import
--connect
jdbc:mysql://localhost:3306/db
--username
root
--password
12345

これを Sqoop のコマンドラインから指定すれば、コマンドラインに入力する手間を省略できる。

bin/sqoop --option-file ./mysql-import.txt

Sqoop は、Windowsのコマンドプロンプトからも同様に使える。

bin\sqoop.cmd help

Sqoop で MySQLのデータを取り込む

まずは、MySQL のデータをHDFSに取り込んでみる。
ここでは、MySQL にサンプルデータベースの sakila があることとしている。
Sqoop は、取り込み処理が、１つの MapReduce タスクなので、あらかじめ、HDFSと YARNが起動していて、HADOOP_HOME 環境変数も設定されている必要がある。

bin/sqoop import --connect jdbc:mysql://localhost:3306/sakila --username root -P --table film -m 1

オプション -P により、実行時に MySQLのパスワードを尋ねるようになる。オプション -m 1 は Mapタスクを1つに限定するもので、こうすると結果にできるファイルが１つになるので、ファイルとして取り込むときには都合がよい。
結果は、HDFS上に MapReduceの結果として残るので、自分のホームにテーブル名のフォルダをみると part-m-00000 のようなファイルとしてある。

$HADOOP_HOME/bin/hdfs dfs -ls film
$HADOOP_HOME/bin/hdfs dfs -get film/part-m-00000

内容は、CSVのテキストデータなので、拡張子を .csv にして、このパスを指定すれば、Drill からクエリできる。

select * from dfs.`c:/tmp/part-m-00000.csv`;

ただ、CSV ファイルでは、データ型がないので、数値もすべて文字列になっている上に、カラム名がないので、Drill からは、配列として扱われるので、多少、不便だろう。

もちろん、Hive上で、対応するテーブルを作成（create table .. row format delimited fields terminated by ',' ..）し、load data inpath .. として、手作業で取り込むことはできるが、最初からそうしたいなら、それは、Sqoop が自動でやってくれる。

Sqoop で MySQL のデータを Hive に取り込む

Hive に取り込めば、テーブルが作られるので、データ型もカラム名もあり、クエリしやすい形になる。
Hive に取り込むには、上記の sqoop のコマンドラインに --hive-import を付け加える。
また、HIVE_HOME 環境変数も設定されている必要がある。

bin/sqoop import --connect jdbc:mysql://localhost:3306/sakila --username root -P --table film --hive-import --hive-table mydb.film

今回は、Hive の metastore がデータファイルを管理してくれるので、複数のファイルに分割されても困らないので、-m 1 オプションを指定していない。
オプション --hive-table mydb.film は、Hive上に mydb データベースがあり、そこに film テーブルとして取り込む指定である。このオプションを指定しないと、default データベースに、MySQLのテーブル名として取り込まれる。

取り込み時の注意としては、Hive で、組み込みメタストアを使っているときは、Hive が起動していると、Sqoop 側からメタストアの操作ができなくてエラーになるので、Hive を停止しておく必要がある。
取り込みに失敗したとき、HDFS上のホームにテーブル名のフォルダが残されたままとなるので、次に Sqoop で、同じ取り込みを指定すると、すでにフォルダがあるとしてエラーになるので、失敗したときには、HDFS上のフォルダを削除することを忘れないようにする。

上記の方法は、Hiveのwarehouse フォルダに取り込まれるが、外部テーブルとして取り込むには、事前に Hive で EXTERNAL TABLE を作成しておき、Sqoop で取り込むときに --hive-table でテーブル名とともに、--target-dir でフォルダ名を指定する。

Hiveとして取り込まれれば、Drill からすばやくクエリできるようになる。

select * from hive.`mydb.film` where rating = "G";

Sqoop で Hiveのデータを MySQLに書き出す

今度は、反対に Hiveのデータを MySQLに書き出してみる。Sqoop は MySQL のテーブルを作成してくれないので、書き出す前に、テーブルが存在している必要がある。
今回は、Hive上にある sample.flightテーブルを書き出す。

Hive上のテーブル定義をみながら、MySQL でテーブルを作成する。

use test;
create table flight (departure varchar(100), arrive varchar(100), year int, month int, day int, delay float, number integer);

Hiveのテーブルの実体は、HDFS上の /user/hive/warehouse/sample.db/flight にある。Sqoop でこのフォルダを指定してMySQLに書き出す。

bin/sqoop export --connect jdbc:mysql://localhost:3306/test --username root -P -m 1 --table flight --export-dir hdfs://localhost:9000/user/hive/warehouse/sample.db/flight --input-fields-terminated-by '\0001'

オプション --export-dir でテーブルの実体を指定するが、HDFS上に存在しているので、hdfs:// から始まるURLとして指定している。また、Hiveのテーブルは、フィールドが '\A' で区切られているので、フィールド区切りを --input-fields-terminated-by で '\0001' として指定している。

Sqoop のその他のオプション

テーブルから取り込むデータを選ぶ

データを取り込むとき、テーブルをそのまま取り込むのではなく、where 句を指定するオプション --where やカラム名を指定するオプション --columns を使って、テーブルの内容を限定できる。

in/sqoop import --connect jdbc:mysql://localhost:3306/sakila --username root -P --table film --columns 'film_id,title,rental_duration,rental_rate' --where 'rating = "G"' --hive-import --hive-table mydb.film3

または、もっと直接的に、 --query オプションを使い、SQLで取り込む内容を指定することもできる。

事前にクエリ内容を確認する

Sqoop は MapReduceタスクが走るので、取り込み結果を事前に確認しておきたいときは、import ではなく eval を使えば、--query オプションのみであるが、Sqoop から、クエリした内容を把握しておける。

bin/sqoop eval --connect jdbc:mysql://localhost:3306/sakila --username root -P --query 'select film_id,title,rental_duration,rental_rate from film where rating = "G"'

ラージオブジェクト

データベース上に、BLOBなどの大きなフィールドがあるとき、これをそのまま取り込むと効率がよくない。Sqoop では、一定サイズ以上のフィールドをサブディレクトリ _lob に別のファイルとして保存する機能があるようで、このサイズを --inline-lob-limit で指定できる。
この機能で省略されたフィールドは、内容としては、外部に保存されたことを示す内容となり、ファイル名とその開始オフセット、長さが同時に記録されるようだ。

ダイレクトモード

MySQL のデータを JDBC経由で取り込むより、ダンプして取り込んだほうが高速なのでは？という疑問は、ダイレクトモードが解決する。
オプション --direct によって、mysqldump を使って取り込みができるようだが、このオプションのときには、制約がいろいろあるようで、ラージオブジェクトのファイル分割ができなかったり、取り込むデータベースによっては使えないなどあるようだ。

定期作業化

テーブルを一気に取り込むほかにも、インクリメンタルに取り込む（UPDATEやINSERTとして）方法や、ジョブとして定期的に取り込むようなやり方もできるようだ。
インクリメンタルに取り込む際には、あるカラムがある値以上の行を取り込むような指定をするなら、オプション --check-column と --last-value で指定できる。
通常は、append モードで、既にある行は更新されない（INSERT）が、オプション --incremental lastmodified によって、更新する（UPDATE）ようにもできる。

定期的に取り込むには、以前に、動作させたワークフローマネージャの Azkaban(http://qiita.com/qtwi/items/e6d3b04c4645e52a0157) も有効に使えるだろう。

今回は、試していないが、SQL ServerやOracle、PostgreSQL などのデータも同様に取り込めるようなので、これで、Drill から統一的にクエリできるデータを増やせるだろう。

以上、お疲れさまでした。