embulk-output-elasticsearchとelasticsearchの注意点 #Embulk

Fluentdのバッチ版Embulk(エンバルク)のまとめ来週のMeetupまでに100になるか気にしている佐藤(@hiroysato)です

この記事はembulkアドベントカレンダー11日目の記事です。

2017/06/29 追記

embulk-output-elasticsearch 4ではメモリリークがあるらしく、大きなデータをロードするとOutOfMemoryになる模様(参考)

2017/5/19 追記

現状

Amazon Elasticsearch Serviceを使う場合embulk-output-elasticsearch_using_url を利用参考
elasticsearch1~5を利用する場合: embulk-output-elasticsearch 4.0以降を利用
AzureやGCPは情報をおまちしております。

2017/3/29 追記

elasticsearch5に対応した、embulk-output-elasticsearch0.4.0がリリースされました。

2017/1/30 追記

elasticsearch 5.0に接続するには、次のプラグインが使えるようです。

2016/11/17 追記

embulk-output-elasticsearch v0.2.Xは、elasticsearch-5.0.0には未対応です。
代わりにembulk-output-elasticsearch_rubyが使えるようです。

2016/1/27 追記

2016年1月26日にembulk-output-elasticsearch v0.2.0が出ました。
ご利用のelasticsearchのバージョンに合わせて適切なプラグインを使ってください。

Elasticsearch	embulkプラグイン
2.X	v0.2.0以降
1.X	v0.1.8

1.Xと2.xで互換性が無いのは、2.0になってJavaのAPIが大きく変わり、下位互換性が失われたためです。

2016/03/08 追加

embulk-output-elasticsearchはAWS ESには未対応です。

@nora96o @matetsu さんの先のツイートで書かれてる通りで、embulk-output-elasticsearchはTransport Client使ってるんですが、AWS ESで使えないので繋がらないです。。 https://t.co/5jKHbNzCdj
— oreradio (@oreradio) 2016年3月8日

以下は古いまとめです。

まとめ

embulk-output-elasticsearch 0.1.8はElasticsearch 2.X系に接続できません。
elasticsearchは1.Xを使いましょう
もし諸般の事情でelasticsearch v2.Xを使いたい場合は下記の方法を試してください。
リリースされたらこのページの情報が変わります。ストックして変更を追跡してください。

はじめに

Embulkのチュートリアルをみて、embulkからelasticsearchへデータを投入してみたがうまくいかない。なぜだとお悩みのかた、
elasticsearchはv2.Xをご利用ではないでしょうか？、

2015年12月11日現在embulk-output-elasticsearch 0.1.8はelasticsearch v2.Xには未対応です。

embulk-output-elasticsearchを利用する場合、Elasticsearch 1.7.3を使ってください。

embulk使ってみたけど動かないじゃんという方が少なくなるといいなと思い

elasticsearch v2.Xに接続した時に出るエラー

elasticsearchのサーバがv2.Xの場合に、embulk runを実行すると次のようにエラーがでます。

これは、リリース済みのembulk-output-elasticsearchがまだv2.Xに未対応のためです。

org.elasticsearch.transport.RemoteTransportException: Failed to deserialize exception response from stream
Caused by: org.elasticsearch.transport.TransportSerializationException: Failed to deserialize exception response from stream
    at org.elasticsearch.transport.netty.MessageChannelHandler.handlerResponseError(MessageChannelHandler.java:176) ~[na:na]
    at org.elasticsearch.transport.netty.MessageChannelHandler.messageReceived(MessageChannelHandler.java:128) ~[na:na]
    at org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) ~[na:na]
    at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) ~[na:na]
    at org.elasticsearch.common.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) ~[na:na]
    at org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:296) ~[na:na]
    at org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462) ~[na:na]
    at org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443) ~[na:na]
    at org.elasticsearch.common.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303) ~[na:na]
    at org.elasticsearch.common.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) ~[na:na]
    at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) ~[na:na]
    at org.elasticsearch.common.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559) ~[na:na]
    at org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:268) ~[na:na]
    at org.elasticsearch.common.netty.channel.Channels.fireMessageReceived(Channels.java:255) ~[na:na]
    at org.elasticsearch.common.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) ~[na:na]
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108) ~[na:na]
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:337) ~[na:na]
    at org.elasticsearch.common.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89) ~[na:na]
    at org.elasticsearch.common.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) ~[na:na]
    at org.elasticsearch.common.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) ~[na:na]
    at org.elasticsearch.common.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) ~[na:na]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_31]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0_31]
    at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_31]
Caused by: java.io.StreamCorruptedException: Unsupported version: 1
    at org.elasticsearch.common.io.ThrowableObjectInputStream.readStreamHeader(ThrowableObjectInputStream.java:46) ~[na:na]
    at java.io.ObjectInputStream.<init>(ObjectInputStream.java:299) ~[na:1.8.0_31]
    at org.elasticsearch.common.io.ThrowableObjectInputStream.<init>(ThrowableObjectInputStream.java:38) ~[na:na]
    at org.elasticsearch.transport.netty.MessageChannelHandler.handlerResponseError(MessageChannelHandler.java:173) ~[na:na]
    ... 23 common frames omitted

embulk-output-elasticsearchを独自ビルド

どうしてもelasticsearch v2.0を試したいという方は、
githubに置いてある、embulk-output-elasticsearchを試してみてください。
githubのソースは、v2.0に対応しているようです。
(ただ不具合があるかもしれませんので、不具合があった場合はIsshueで報告をしてください)

こちらをビルドして試すことができます。(※JDKが必要です。)

次のコマンドで、elasticsearch v2.X対応のembulk-output-elasticsearchを作ることができます。

git clone https://github.com/muga/embulk-output-elasticsearch.git
cd embulk-output-elasticsearch
./gradlew gem

検証

elasticsearch導入

wget https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/zip/elasticsearch/2.1.0/elasticsearch-2.1.0.zip
unzip elasticsearch-2.1.0.zip 
cd elasticsearch-2.1.0
./bin/elasticsearch

サンプルデータの作成

embulk example sample
embulk guess sample/example.yml -o config.yml

設定ファイルの修正

outの部分を変更します。

in:
  type: file
  path_prefix: /private/tmp/sample/csv/sample_
  decoders:
  - {type: gzip}
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    trim_if_not_quoted: false
    skip_header_lines: 1
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name: id, type: long}
    - {name: account, type: long}
    - {name: time, type: timestamp, format: '%Y-%m-%d %H:%M:%S'}
    - {name: purchase, type: timestamp, format: '%Y%m%d'}
    - {name: comment, type: string}
#out: {type: stdout}
out:
  type: elasticsearch
  index: embulk
  index_type: embulk
  nodes:
    - host: localhost

プレビュー

embulk preview config.yml
+---------+--------------+-------------------------+-------------------------+----------------------------+
| id:long | account:long |          time:timestamp |      purchase:timestamp |             comment:string |
+---------+--------------+-------------------------+-------------------------+----------------------------+
|       1 |       32,864 | 2015-01-27 19:23:49 UTC | 2015-01-27 00:00:00 UTC |                     embulk |
|       2 |       14,824 | 2015-01-27 19:01:23 UTC | 2015-01-27 00:00:00 UTC |               embulk jruby |
|       3 |       27,559 | 2015-01-28 02:20:02 UTC | 2015-01-28 00:00:00 UTC | Embulk "csv" parser plugin |
|       4 |       11,270 | 2015-01-29 11:54:36 UTC | 2015-01-29 00:00:00 UTC |                       NULL |
+---------+--------------+-------------------------+-------------------------+----------------------------+

データ投入

独自ビルドしたembulk-output-elasticsearchのパスを指定してembulk runを動かします。

embulk run -I /path/to/embulk-output-elasticsearch/lib config.yml

ビルドしたパッケージをインストールする場合

次の方法でインストールできます。(バージョンが0.1.8になるので注意)

./gradlew gem
embulk gem install -l pkg/embulk-output-elasticsearch-0.1.8.gem