Java で Embulk Output Plugin を書く

はじめに

すでに Ruby で書かれた Embulk plugins が登場しており、Embulk の利用シーンも段々と増えていると感じます。その一方で、Java で書かれた plugins はまだまだ少ないのではないかと思います。ちょうど、Embulk から Elasticsearch へデータを書き出す output plugin を書く機会があったので、その plugin の書き方をメモとして残します。

まずは Plugin Template を利用

Ruby であれ Java であれ plugins を作成するには、まず Embulk に用意されている new コマンドを実行するのがよいかと思います。New コマンドは、Plugin 開発に必要な gradle スクリプトや Embulk 本体が plugins を登録するために必要なスクリプトなどを自動で生成します。Embulk ではどの category (output, input, filter, etc.) の plugin を開発するかで、Embulk を拡張するポイントが異なりますが、その拡張のひな形も自動で生成します。

今回は Java で Elasticsearch 書き出し用の output plugin を作成するので、category として 'java-output'、name に 'elasticsearch' を指定して実行します。

$ java -jar embulk.jar new java-output elasticsearch

Plugin Type の宣言を確認

上記の new コマンドを実行することで、lib/embulk/output/elasticsearch.rb というファイルが作成されましたが、これは Embulk 本体が起動時に自分が利用できる plugin として登録するために必要な記述です。

elasticsearch.rb

Embulk::JavaPlugin.register_output(
  :elasticsearch, "org.embulk.output.ElasticsearchOutputPlugin",
  File.expand_path('../../../../classpath', __FILE__))

この宣言を Embulk 本体が読むことで、以下のような 'elasticsearch' という type を利用者は設定ファイルに指定することができます。

config.yml

out:
  type: elasticsearch

Embulk::JavaPlugin クラスとは Embulk 本体の lib/embulk/java_plugin.rb です。ここには、register_input や register_output などの class methods が定義されています。Java の output plugin を Embulk に登録するには、register_output を使います。Plugin type は :elasticsearch として symbol で宣言し、実際の plugin の実装である Java のクラス名を同時に渡します。

Output Plugin を Java で実装

New コマンドはまた 'org.embulk.output.ElasticsearchOutputPlugin' というクラスを作成しますが、ここに Java で output plugin を実装します。Java で output plugin クラスを書くには、そのクラスは embulk-core に含まれている org.embulk.spi.OutputPlugin を実装する必要があります。現在の Elasticsearch output plugin は少しさぼっており、resume と cleanup メソッドが実装されていません。少なくとも、resume オプションなしで Embulk を実行できるようにするため、transaction と open メソッドを実装しました。ここで実装する transaction メソッドは、Embulk の LocalExecutor#doRun メソッドの中で呼び出されます。一方、open は、LocalExecutor#startProcessor の中で使われます。

Open メソッドの戻り値の型は org.embulk.spi.TransactionalPageOutput というもので、このオブジェクトを表現しているクラスに、Embulk から渡されるデータをどこに出力させるかを定義します。特に、Embulk は TransactionalPageOutput の add(Page) メソッドを呼び出し、データを output plugin 側に渡します。この渡された Page オブジェクトは複数レコードの集まりです。これを PageReader を使って、1 レコードずつ読み出し、どこかへ出力させます。Elasticsearch には、bulk api が用意されており、クライアントは複数レコードの indexing をまとめて 1 リクエストで行えます。Elasticsearch output plugin ではこれを利用しています。

また、Embulk の起動時に読み込む設定ファイルにデータを出力させる elasticsearch の cluster 名等をオプションで書けた方が便利です。これには、org.embulk.config.Task を実装したクラスを宣言します。

Java Plugin の gem を作成

Embulk の plugins は実装言語に依らず、gem で配布されます。rubygems.org で公開されている gem の Embulk plugins を、Embulk は簡単に取得し、利用することができます。本メモの最初に説明した new コマンドを利用すると、build.gradle に gem の作成タスクが自動で宣言されます。これにより gradle を実行することで、gem ファイルが作成されます。

$ ./gradlew gem

Plugins を Embulk 本体へ登録

すでに Embulk-plugin-inputの作り方 等のページで、Embulk の plugin の読み込ませ方が記載されています。

開発時には試行錯誤がつきものですので、Embulk plugin 開発時に Embulk 本体へ手早く開発中の plugin を読み込ませ動作確認をしたいときがあります。Embulk の run コマンドの -I オプションで、その plugin の gem ファイルまで作成せずに Embulk 本体にその plugin を読み込ませテストすることができます。

$ java -jar embulk.jar run -I <path-to-plugin>/lib config.yml

ただし、上記のコマンドを繰り返し実行する前に、./gradlew classpath や ./gradle gem 等を利用し、あらかじめ classpath/ にプラグインに依存関係のある jar files を取得しておく必要があります（Feb 16, 2015 追記）。

上記のように、Java plugins の開発・テストができます。Ruby だけではなく、Java でも Embulk plugins がどんどん実装されていくとうれしいですね。また、Embulk に Java plugins を読み込ませる方法は他にもありますが、機会があればメモ書きしようと思います。