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

概要

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のインストール

MySQLをインストールしたら、以下の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のインストール手順と簡単な使い方

javaのインストール

以下を参考にしてインストールしてください
CentOS7にJava OpenJDK8のインストール

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するため、以下のプラグインを入れます。

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の接続情報を、環境変数としてセットします。

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に、以下のように記述する方法でも構いません。

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 }}のような形式で使います。
ファイル名は以下の通りでなく、任意のもので構いません。

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

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

{
  "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系クラスタ構成構築手順

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

概要

環境