EmbulkでMySQLのデータをElasticsearch5に転送する

## 概要
Embulkを使って、MySQLのデータをElasticsearch5に転送します。

## 環境
- CentOS 7.2
- embulk 0.8.35
- openjdk version "1.8.0_144"
- MySQL 5.7
- elasticsearch 5.6.3-1

## 事前準備
### MySQLのインストール
以下を参考にして、mysqlをインストールしてください。
[CentOS6 or 7にMySQL5.6 or 5.7のインストール](https://qiita.com/Esfahan/items/83200c64de8d826677b5)

MySQLをインストールしたら、以下のsql文でテーブルを作成してデータを投入してください。

```sql
GRANT ALL ON sample_db.* TO sample_user@'%' IDENTIFIED BY 'Sample_password123';
CREATE DATABASE sample_db;
USE sample_db;

CREATE TABLE items(
    id INT NOT NULL AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(20),
    price INT(10),
    created_date TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_date TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

INSERT INTO items (name, price) VALUES('olive oil', 600);
INSERT INTO items (name, price) VALUES('garlic', 300);
INSERT INTO items (name, price) VALUES('red peppar', 200);
```

### Elasticsearchのインストール
以下を参考にして、Elasticsearchをインストールしてください。
[Elasticsearchのインストール手順と簡単な使い方](https://qiita.com/Esfahan/items/963225cd2d2b6b219c05)


### javaのインストール
以下を参考にしてインストールしてください
[CentOS7にJava OpenJDK8のインストール](https://qiita.com/Esfahan/items/60cf425514c66553bd42)


## Embulkのインストール
以下のコマンドで、Embulkをインストールします。

```
$ curl --create-dirs -o ~/.embulk/bin/embulk -L "https://dl.embulk.org/embulk-latest.jar"
$ chmod +x ~/.embulk/bin/embulk
$ echo 'export PATH="$HOME/.embulk/bin:$PATH"' >> ~/.bashrc
$ source ~/.bashrc
```

インストール確認

```
$ embulk -version
embulk 0.8.35
```

## ディレクトリ構成
sample-appというディレクトリを作成し、以下の作業は全てその中で行います。

```
$ mkdir sample-app
```

## pluginの管理
Embulkのpluginを管理するために、mkbundleというものをインストールします。
railsのbundlerみたいなものです。
以下のコマンドを実行してください。

```
$ cd sample-app
$ embulk mkbundle sample-plugins
```

### Gemfileにpluginを追加

今回はMySQLからinputして、elasticsearch5にoutputするため、以下のプラグインを入れます。

```rb:sample-app/sample-plugins/Gemfile
gem 'embulk-input-mysql'
gem 'embulk-output-elasticsearch_ruby'
```

### pluginをインストール
railsでいうbundle install。
Gemfileのある階層で実行します。

```
$ cd sample-plugins
$ embulk bundle --path=vendor/bundle
```

### pluginがインストールされたことを確認

```
$ embulk bundle list

2017-10-18 07:10:53.791 +0000: Embulk v0.8.35
Gems included by the bundle:
  * bundler (1.10.6)
  * elasticsearch (5.0.4)
  * elasticsearch-api (5.0.4)
  * elasticsearch-transport (5.0.4)
  * embulk (0.8.35)
  * embulk-input-mysql (0.8.5)
  * embulk-output-elasticsearch_ruby (0.1.6)
  * excon (0.59.0)
  * faraday (0.13.1)
  * liquid (4.0.0)
  * msgpack (1.1.0)
  * multi_json (1.12.2)
  * multipart-post (2.0.0)
  * rjack-icu (4.54.1.1)
```

## 環境変数
sample-appの直下に、以下を作成してください。
MySQL、Elasticsearchの接続情報を、環境変数としてセットします。

```bash:sample-app/.env
MYSQL_HOST=localhost
MYSQL_USER=sample_user
MYSQL_PASSWORD=Sample_password123
ELS_HOST=localhost
```

```
$ export $(cat .env | xargs)
```

~/.bash_profileとか、/etc/environmentに、以下のように記述する方法でも構いません。

```bash
export MYSQL_HOST=localhost
export MYSQL_USER=sample_user
export MYSQL_PASSWORD=Sample_password123
export ELS_HOST=localhost
```

```
$ source ~/.bash_profile
```

## yamlファイルを作成
liquid拡張子にすることで、yamlファイル内で、include文や、変数を使うことができるようになります。
ちなみに、環境変数は、`{{ env.MYSQL_USER }}`のような形式で使います。
ファイル名は以下の通りでなく、任意のもので構いません。

```yaml:sample-app/mysql-elasticsearch.yml.liquid
in:
  type: mysql
  host: {{ env.MYSQL_HOST }}
  user: {{ env.MYSQL_USER }}
  password: {{ env.MYSQL_PASSWORD }}
  database: sample_db
  query: SELECT * FROM items

out:
  type: elasticsearch_ruby
  # 省略可能
  cluster_name: sample-application
  index: sample_index
  index_type: sample_type
  nodes:
    - host: {{ env.ELS_HOST }}
```

## 実行
実行する前に、dry-runをします。

```
$ embulk preview mysql-elasticsearch.yml.liquid -b sample-plugins

+---------+-------------+------------+-------------------------+-------------------------+
| id:long | name:string | price:long |  created_date:timestamp |  updated_date:timestamp |
+---------+-------------+------------+-------------------------+-------------------------+
|       1 |   olive oil |        600 | 2017-10-18 07:07:36 UTC | 2017-10-18 07:07:36 UTC |
|       2 |      garlic |        300 | 2017-10-18 07:07:36 UTC | 2017-10-18 07:07:36 UTC |
|       3 |  red peppar |        200 | 2017-10-18 07:07:36 UTC | 2017-10-18 07:07:36 UTC |
+---------+-------------+------------+-------------------------+-------------------------+
```

問題がなければ、実行します。

```
$ embulk run mysql-elasticsearch.yml.liquid -b sample-plugins/
```

## Elasticsearchにデータが入ったか確認
下記のコマンドで、Elasticsearchに問い合わせます。

```
$ curl http://localhost:9200/embulk_sample/embulk/_search?pretty
```

以下のような結果が返ってくれば、成功です。

```json
{
  "took" : 154,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 3,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "embulk_sample",
        "_type" : "embulk",
        "_id" : "AV8t1p0iQ_tnL6hHKk6O",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "garlic",
          "price" : 300,
          "created_date" : "2017-10-18 11:27:42 UTC",
          "updated_date" : "2017-10-18 11:27:42 UTC"
        }
      },
      {
        "_index" : "embulk_sample",
        "_type" : "embulk",
        "_id" : "AV8t1p0iQ_tnL6hHKk6N",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "olive oil",
          "price" : 600,
          "created_date" : "2017-10-18 11:26:43 UTC",
          "updated_date" : "2017-10-18 11:26:43 UTC"
        }
      },
      {
        "_index" : "embulk_sample",
        "_type" : "embulk",
        "_id" : "AV8t1p0iQ_tnL6hHKk6P",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "red peppar",
          "price" : 200,
          "created_date" : "2017-10-18 11:28:08 UTC",
          "updated_date" : "2017-10-18 11:28:08 UTC"
        }
      }
    ]
  }
}
```

以上

## 関連記事
- [Elasticsearch5系 クラスタ構成構築手順](https://qiita.com/Esfahan/items/3c07bfbb57c7098e9531)

## 参考
- [Github: embulk](https://github.com/embulk/embulk#quick-start)
- [大量データの転送にEmbulkを使ってみたら本当に楽だった](http://techlog.voyagegroup.com/entry/2017/07/31/173839)
- [Embulkではてなブログの記事をparseしてElasticsearchに入れてみる](http://yudoufu.hatenablog.jp/entry/2015/06/11/083042)
- [embulk mkbundle の使い方](https://qiita.com/sonots/items/979c8b0810ccffc47317)
- [embulk チュートリアルを兼ねて RDS のスロークエリをログを Elasticsearch と Kibana で可視化する](http://inokara.hateblo.jp/entry/2016/05/08/101816)
- [ApacheのアクセスログをEmbulk+Elasticsearch+Kibanaで見る](https://qiita.com/segawa/items/9c0e549219f48269529f)