更新日: 2014年5月19日
Elasticsearch

ElasticsearchにMySQLからデータ挿入、JDBC River Pluginのインストールと使い方

MySQL から Elasticsearch へデータ取り込みを行う JDBC River Plugin をインストールします。以下エントリーの続きとなります。

ElasticsearchのインストールとCSVからのデータ挿入 | EasyRamble
Elasticsearchのクエリとフィルターで簡単な検索を試す例 | EasyRamble

【お知らせ】英単語を画像イメージで楽に暗記できる辞書サイトを作りました。英語学習中の方は、ぜひご利用ください！

画像付き英語辞書 Imagict | 英単語をイメージで暗記
【開発記録】
英単語を画像イメージで暗記できる英語辞書サービスを作って公開しました

— 環境 —
Mac OS X Lion 10.7.5

JDBC driver for MySQL のダウンロード

JDBC River Plugin の動作には、Java から MySQL への接続するための JDBC driver が必要なので、あらかじめ利用するDBベンダーの JDBC driver をダウンロードしておきます。私は、MySQL 利用なので以下のページからダウンロードしました。

MySQL :: Download Connector/J

ダウンロードしたら解凍して、.jar ファイルを /usr/share/java/ に移動します。


$ cd ~/Downloads
$ tar xvfz mysql-connector-java-5.1.30.tar.gz
$ cd mysql-connector-java-5.1.30
$ sudo mv mysql-connector-java-5.1.30-bin.jar /usr/share/java

$ cd ~/Downloads

$ tar xvfz mysql-connector-java-5.1.30.tar.gz

$ cd mysql-connector-java-5.1.30

$ sudo mv mysql-connector-java-5.1.30-bin.jar /usr/share/java

その後、MySQL JDBC driver へのパスを CLASSPATH 環境変数に追加。


$ vi ~/.zshrc
# MySQL JDBC driver
export CLASSPATH=$CLASSPATH:/usr/share/java/mysql-connector-java-5.1.30-bin.jar

$ vi ~/.zshrc

# MySQL JDBC driver

export CLASSPATH=$CLASSPATH:/usr/share/java/mysql-connector-java-5.1.30-bin.jar

変更を反映させる。


$ source ~/.zshrc

$ source ~/.zshrc

Java から MySQL への接続確認

テスト用の sample データベースを作成しておきます。


$ mysql -u root -p
mysql> create database sample;

$ mysql -u root -p

mysql> create database sample;

Java から MySQL への接続を確認するために簡単な Java プログラムを作成します。説明のないとってもシンプルなサンプルプログラム集からお借りしました。


$ cd ~/work
$ vi mysql.java
import java.io.*;
import java.net.*;
import java.sql.*;
class Test{
  public static void main(String args[]){
    try {
      Class.forName("com.mysql.jdbc.Driver");
      Connection con = DriverManager.getConnection(
          "jdbc:mysql://localhost/sample","root","12345678");
      Statement st = con.createStatement();
      st.executeUpdate("CREATE TABLE mytbl (id INT,name VARCHAR(10))");
      st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (1,'apple')");
      st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (2,'banana')");
      st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (3,'candy')");
      ResultSet rs = st.executeQuery("SELECT id,name FROM mytbl");
      while(rs.next()){
        System.out.println(rs.getString("id") +" "+ rs.getString("name"));
      }
      st.executeUpdate("DROP TABLE mytbl");
      con.close();
    } catch (Exception e) {
      System.out.println("ERROR:" + e );
    }
  }
}

$ cd ~/work

$ vi mysql.java

import java.io.*;

import java.net.*;

import java.sql.*;

class Test{

public static void main(String args[]){

try {

Class.forName("com.mysql.jdbc.Driver");

Connection con = DriverManager.getConnection(

"jdbc:mysql://localhost/sample","root","12345678");

Statement st = con.createStatement();

st.executeUpdate("CREATE TABLE mytbl (id INT,name VARCHAR(10))");

st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (1,'apple')");

st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (2,'banana')");

st.executeUpdate("INSERT INTO mytbl (id,name) VALUES (3,'candy')");

ResultSet rs = st.executeQuery("SELECT id,name FROM mytbl");

while(rs.next()){

System.out.println(rs.getString("id") +" "+ rs.getString("name"));

}

st.executeUpdate("DROP TABLE mytbl");

con.close();

} catch (Exception e) {

System.out.println("ERROR:" + e );

}

12345678 の部分は、MySQL の root ユーザーのパスワードに変更します。あとは、コンパイルして実行。


$ javac mysql.java
$ java Test
1 apple
2 banana
3 candy

$ javac mysql.java

$ java Test

1 apple

2 banana

3 candy

無事に、MySQL への接続を確認できました。これで準備OK。

JDBC River Plugin をインストール

ここからが本番ですが、MySQL 内のデータを Elasticsearch にインデックスさせるために、Elasticsearch のプラグインである JDBC River Plugin をインストールします。様々な River Plugin が用意されているので、SQL 以外のデータ形式からも色々 Elasticsearch に取り込めるようです。

https://github.com/jprante/elasticsearch-river-jdbc

JDBC River Plugin をインストール。


$ plugin --install jdbc --url http://xbib.org/repository/org/xbib/elasticsearch/plugin/elasticsearch-river-jdbc/1.1.0.1/elasticsearch-river-jdbc-1.1.0.1-plugin.zip

$ plugin --install jdbc --url http://xbib.org/repository/org/xbib/elasticsearch/plugin/elasticsearch-river-jdbc/1.1.0.1/elasticsearch-river-jdbc-1.1.0.1-plugin.zip

Github の readme にある通りに、MySQL JDBC driver を $ES_HOME/plugins/jdbc/ にコピーします。$ES_HOME がどこか分からなかったので、まずは plugins のディレクトリを調べました。


$ brew info elasticsearch
elasticsearch: stable 1.1.1, HEAD
http://www.elasticsearch.org
/usr/local/Cellar/elasticsearch/1.1.1 (43 files, 22M) *
  Built from source
From: https://github.com/Homebrew/homebrew/commits/master/Library/Formula/elasticsearch.rb
==> Caveats
Data:    /usr/local/var/elasticsearch/elasticsearch_*****/
Logs:    /usr/local/var/log/elasticsearch/elasticsearch_*****.log
Plugins: /usr/local/var/lib/elasticsearch/plugins/
...

$ brew info elasticsearch

elasticsearch: stable 1.1.1, HEAD

http://www.elasticsearch.org

/usr/local/Cellar/elasticsearch/1.1.1 (43 files, 22M) *

Built from source

From: https://github.com/Homebrew/homebrew/commits/master/Library/Formula/elasticsearch.rb

==> Caveats

Data: /usr/local/var/elasticsearch/elasticsearch_*****/

Logs: /usr/local/var/log/elasticsearch/elasticsearch_*****.log

Plugins: /usr/local/var/lib/elasticsearch/plugins/

...

driver をコピー。


$ cp /usr/share/java/mysql-connector-java-5.1.30-bin.jar /usr/local/var/lib/elasticsearch/plugins/jdbc

$ cp /usr/share/java/mysql-connector-java-5.1.30-bin.jar /usr/local/var/lib/elasticsearch/plugins/jdbc

MySQL にテスト用テーブルとデータを用意して試す

テスト用の MySQL データとして以下を用意しました。


mysql> create table test(id integer, name varchar(20));
mysql> insert into test (id, name) values (1, "りんご");
mysql> insert into test (id, name) values (2, "リンゴ");
mysql> insert into test (id, name) values (3, "林檎");
mysql> insert into test (id, name) values (4, "apple");
mysql> select * from test;
+------+-----------+
| id   | name      |
+------+-----------+
|    1 | りんご    |
|    2 | リンゴ    |
|    3 | 林檎      |
|    4 | apple     |
+------+-----------+
4 rows in set (0.00 sec)

mysql> create table test(id integer, name varchar(20));

mysql> insert into test (id, name) values (1, "りんご");

mysql> insert into test (id, name) values (2, "リンゴ");

mysql> insert into test (id, name) values (3, "林檎");

mysql> insert into test (id, name) values (4, "apple");

mysql> select * from test;

+------+-----------+

| id | name |

+------+-----------+

| 1 | りんご |

| 2 | リンゴ |

| 3 | 林檎 |

| 4 | apple |

+------+-----------+

4 rows in set (0.00 sec)

このデータを Elasticsearch に取り込んでみます。


$ curl -XPUT 'localhost:9200/_river/my_jdbc_river/_meta' -d '{
    "type" : "jdbc",
    "jdbc" : {
        "url" : "jdbc:mysql://localhost:3306/sample",
        "user" : "root",
        "password" : "*****",
        "sql" : "select * from test"
    }
}'

$ curl -XPUT 'localhost:9200/_river/my_jdbc_river/_meta' -d '{

"type" : "jdbc",

"jdbc" : {

"url" : "jdbc:mysql://localhost:3306/sample",

"user" : "root",

"password" : "*****",

"sql" : "select * from test"

}

elasticsearch を再起動。http://127.0.0.1:9200/_plugin/head/ にアクセスすると、MySQL からデータが取り込まれたのを確認できました。しかも CSV River の時と違って、JDBC River では日本語が文字化けせずに正常にインポートできています！

以上の方法では事前に Elasticsearch で mapping を作成せずに、MySQL → Elasticsearch へと直接データを入れたので、Elasticsearch のフィールドの型などは動的に決定されることになります。head プラグインで動的に作成された mappings を確認すると以下のように、id は long 型、name は sgring 型になっていました。


mappings: {
    jdbc: {
        properties: {
            id: {
                type: long
            }
            name: {
                type: string
            }
        }
    }
}

mappings: {

jdbc: {

properties: {

id: {

type: long

}

name: {

type: string

}

mapping が動的に作成されるのは若干怖い感じがしますので、以降では、事前に mapping を作成してその後 JDBC River プラグインを使う方法を紹介します。

事前に mapping を作成して MySQL からインポート

次に、Elasticsearch 側で事前に以下の mapping を作成してデータのインポートを試してみます。index 名は sample。type 名は test。


$ curl -XPOST localhost:9200/sample -d '
{
  "settings": {
    "analysis": {
      "tokenizer": {
        "ngram_tokenizer": {
          "type": "nGram",
          "min_gram": "2",
          "max_gram": "3",
          "token_chars": [
            "letter",
            "digit"
          ]
        }
      },
      "analyzer": {
        "ngram_analyzer": {
          "tokenizer": "ngram_tokenizer"
        }
      }
    }
  },
  "mappings": {
    "test": {
      "_source": {
        "enabled": true
      },
      "_all": {
        "enabled": true,
        "analyzer": "ngram_analyzer"
      },
      "properties": {
        "id": {
          "type": "integer",
          "index": "not_analyzed"
        },
        "name": {
          "type": "string",
          "index": "analyzed",
          "analyzer": "ngram_analyzer"
        }
      }
    }
  }
}'

{"acknowledged":true}%

$ curl -XPOST localhost:9200/sample -d '

{

"settings": {

"analysis": {

"tokenizer": {

"ngram_tokenizer": {

"type": "nGram",

"min_gram": "2",

"max_gram": "3",

"token_chars": [

"letter",

"digit"

]

}

"analyzer": {

"ngram_analyzer": {

"tokenizer": "ngram_tokenizer"

}

"mappings": {

"test": {

"_source": {

"enabled": true

"_all": {

"enabled": true,

"analyzer": "ngram_analyzer"

"properties": {

"id": {

"type": "integer",

"index": "not_analyzed"

"name": {

"type": "string",

"index": "analyzed",

"analyzer": "ngram_analyzer"

}

{"acknowledged":true}%

この作成した index/type => sample/test に MySQL からデータを取り込んでみます。jdbc プロパティで、index に sample、type に test を指定します。


$ curl -XPUT 'localhost:9200/_river/my_jdbc_river/_meta' -d '{
    "type" : "jdbc",
    "jdbc" : {
        "url" : "jdbc:mysql://localhost:3306/sample",
        "user" : "root",
        "password" : "*****",
        "sql" : "select * from test",
        "index" : "sample",
        "type" : "test"
    }
}'

$ curl -XPUT 'localhost:9200/_river/my_jdbc_river/_meta' -d '{

"type" : "jdbc",

"jdbc" : {

"url" : "jdbc:mysql://localhost:3306/sample",

"user" : "root",

"password" : "*****",

"sql" : "select * from test",

"index" : "sample",

"type" : "test"

}

http://127.0.0.1:9200/_plugin/head/ で確認すると、sample/test にデータが無事に取り込まれました。このように先に mapping を定義し index/type を作成した後に、JDBC River で index/type を指定してデータをインポートしてやるとフィールドの型を独自に自由に決められます。個人的にはこちらの方法のほうが安心です。

JDBC River Plugin を用いた MySQL から Elasticsearch へのデータ取り込みは以上です。