embulk-filter-copy の紹介

# はじめに
[embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) という [Embulk Plugin](http://www.embulk.org/plugins/) を作って[リリース](https://rubygems.org/gems/embulk-filter-copy)しました。
この記事では、その使い方や実装コンセプトについてまとめようと思います。

# なにするやつか

filter plugin に渡ってきたデータを copy して、そのデータを input として別の embulk を実行する plugin です。example を見ると雰囲気が伝わると思います。

```config.yml
in:
  type: file
  path_prefix: ./example/data.tsv
  parser:
    type: csv
    delimiter: "\t"
    skip_header_lines: 0
    null_string: ""
    columns:
      - { name: id, type: long }
      - { name: description, type: string }
      - { name: name, type: string }
      - { name: payload, type: json}
    stop_on_invalid_record: true
filters:
  - type: copy
    config:
      exec:
        max_threads: 8
      filters:
        - type: remove_columns
          remove: ["id"]
      out:
        type: stdout
out:
  type: stdout

```

`type: copy` の下あたりに embulk の config っぽい記述がありますね！
例えば、処理するデータが以下のようなデータだったとすると (4行目の3カラム目は空文字)

```example/data.tsv
0	c20ef94602	c212c89f91	{"a":0,"b":"99"}
1	330a9fc33a	e25b33b616	{"a":1,"b":"a9"}
2	707b3b7588	90823c6a1f	{"a":2,"b":"96"}
3	8d8288e66f		{"a":3,"b":"86"}
4	c54d8b6481	e56a40571c	{"a":4,"b":"d2"}
```
出力は次のようになります。(ログは省略)

```stdout
0,c20ef94602,c212c89f91,{"a":0,"b":"99"}
1,330a9fc33a,e25b33b616,{"a":1,"b":"a9"}
2,707b3b7588,90823c6a1f,{"a":2,"b":"96"}
3,8d8288e66f,,{"a":3,"b":"86"}
4,c54d8b6481,e56a40571c,{"a":4,"b":"d2"}
c20ef94602,c212c89f91,{"a":0,"b":"99"}
330a9fc33a,e25b33b616,{"a":1,"b":"a9"}
707b3b7588,90823c6a1f,{"a":2,"b":"96"}
8d8288e66f,,{"a":3,"b":"86"}
c54d8b6481,e56a40571c,{"a":4,"b":"d2"}
```

main の Embulk の`stdout`と [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) が実行する Embulk の`stdout`(`id`カラムが除かれている方) が出力されていることが分かると思います。

このように input 以外の embulk 実行設定を [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) の config に渡すことで [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) に渡ってきたデータを input として embulk を別途起動するプラグインです。

# つかいかた
[なにするやつか](http://qiita.com/Civitaspo/items/da8483c28817071d90dc#%E3%81%AA%E3%81%AB%E3%81%99%E3%82%8B%E3%82%84%E3%81%A4%E3%81%8B) でほぼ説明してしまっていますが、 [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) は以下のオプションを持っています。

- `config`: `in:` を除いた [embulk の config](http://www.embulk.org/docs/built-in.html#embulk-configuration-file-format) を書きます。

これら以外に必要なオプションはありません。

# 利用に際しての注意点
### Java8
- [EmbulkとJava7](http://qiita.com/hiroysato/items/7b5f77c6ef752c4cd9ca) という記事で紹介されていますが 2017-06-09現在 Embulk はまだ Java8 を公式にサポートしていません:scream:
- しかし、 [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) は Java8 で実装されています。
- コンパイルも Java7 向けのコンパイルをしていません。(ラムダ式などjava7向けにコンパイルできないJava8の機能を使っています。)
- これは後に紹介する依存ライブラリが Java8 前提で書かれていたことによる制約です。 ~~趣味で書いたから楽しみたかった~~
- Embulk は Runtime を Java8 にすることに関して問題はない（はず）ので、ご利用の際は Java8 による実行をお願いします。
- Embulk の Java8 公式サポート切に願っております:pray:よろしくお願いいたします:pray::pray::pray:

### Interface
- Embulk の config を渡す `config` というオプションについては変更することは今後もありません:+1:(あったとしても一定期間並行利用できるようにします。)
- ただ、実は [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) には紹介したオプション以外のオプションが多数あります。(まだ README にも書いてません)
- これらは内部で利用しているライブラリを細かく制御できるようにするためのものですが、内部実装は利用が進むと変わる可能性があるため、これらオプションについては変わる可能性があることをご認識ください。

### エラーハンドリング
- まだ仕様が固まっていません:poop:
- 現状は、 [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) 上で動く Embulk が Exception を吐いた場合は問答無用で [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) も RuntimeException を吐く（つまり、 main で動いている Embulk がエラーで死ぬ）ようにしています。
- 今後、例えば `skip_if_embulk_failed` といったオプションを追加して Exception 無視出来るようにしたりするアイデアはありますが、利用が進んで利用者の声を聞きつつ進めようと思っています。
- ので、こうあるべきという意見があれば [issue](https://github.com/civitaspo/embulk-filter-copy/issues) に起票をお願いします:pray: 

### [embulk-executor-mapreduce](https://github.com/embulk/embulk-executor-mapreduce)
- [embulk-executor-mapreduce](https://github.com/embulk/embulk-executor-mapreduce) 上ではまだ動きません。

### その他
- [README の NOTE](https://github.com/civitaspo/embulk-filter-copy#note) を参照してください。
- まだ細かな配慮が出来ていない部分があるのでお気づきの点があれば [issue](https://github.com/civitaspo/embulk-filter-copy/issues) に起票をお願いします:pray:

# モチベーション
[Embulk Meetup Tokyo #3](https://eventdots.jp/event/618872) で @toyama0919 さんの[『クラウドとEmbulkで作った機械学習基盤』](https://www.slideshare.net/toyama0919/embulk-and-machine-learning-infrastructure)という発表を見て、多段かつ並列に [Embulk](http://www.embulk.org/) を実行してデータ生成をする複雑なユースケースを知り [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) を実装してみようと思いました。その複雑さについてはスライドや以下の tweet 近辺をご覧になってみてください笑
<blockquote class="twitter-tweet" data-lang="en-gb"><p lang="ja" dir="ltr">Embulk と digdag で map reduce やってるかんじかｗ <a href="https://twitter.com/hashtag/embulk?src=hash">#embulk</a></p>&mdash; Civitaspo (@Civitaspo) <a href="https://twitter.com/Civitaspo/status/864430583908229120">16 May 2017</a></blockquote>
<script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>
<blockquote class="twitter-tweet" data-lang="en-gb"><p lang="ja" dir="ltr">toyamaさんembulkで相当無茶やってる気がするw <a href="https://twitter.com/hashtag/embulk?src=hash">#embulk</a></p>&mdash; 夜行性のフレンズ じょーかー (@joker1007) <a href="https://twitter.com/joker1007/status/864430006121897984">16 May 2017</a></blockquote>
<script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

2017-06-07 に開催された [workflowenginesnight](http://labotech.dmm.com/entry/2017/06/08/173758) では @frsyuki さんに「ユースケースの幅が広がるので良いですねぇ」とお墨付きの言葉をもらったのでそれも今ではモチベーションになっています。

# 実装コンセプト
ここからは実装コンセプトについて書いていきます。

**言葉遣いについてのおことわり**
Embulk と書いたときに [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) が実行した Embulk か main で実行されている Embulk なのかわかりにくいので、ここからは main で実行されている Embulk を main Embulk、 [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) が実行した Embulk を copy Embulk と書くことにします。

### [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) は copy Embulk が使う input plugin も実装する

[embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) は以下のコードを見ると分かりますが [input plugin も実装](https://github.com/civitaspo/embulk-filter-copy/blob/v0.0.1/lib/embulk/filter/copy.rb) されています。

```lib/embulk/filter/copy.rb
Embulk::JavaPlugin.register_filter(
  "copy", "org.embulk.filter.copy.CopyFilterPlugin",
  File.expand_path('../../../../classpath', __FILE__))

Embulk::JavaPlugin.register_input(
    "internal_forward",
    "org.embulk.filter.copy.plugin.InternalForwardInputPlugin",
    File.expand_path('../../../../classpath', __FILE__))
```

これは [copy Embulk が使う input plugin](https://github.com/civitaspo/embulk-filter-copy/blob/v0.0.1/src/main/java/org/embulk/filter/copy/plugin/InternalForwardInputPlugin.java) で [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) がコピーしたデータを受信するために使っています。

### main Embulk から copy Embulk へのデータ送信はネットワーク経由で行う
main Embulk から copy Embulk へのデータ送信はネットワーク経由で行います。これは、[将来的に別のサーバーで copy Embulk を起動すること](https://github.com/civitaspo/embulk-filter-copy/issues/6)を考えているためです。Embulk はサーバーリソースを限界まで使うよう設計されているため copy Embulk は別のサーバーで動くのが良いだろうと考えています。

##### [embulk-executor-mapreduce](https://github.com/embulk/embulk-executor-mapreduce)
そのため、[embulk-executor-mapreduce](https://github.com/embulk/embulk-executor-mapreduce) 上で動かすことも将来的に出来ると考えています。copy Embulk の input task が動くサーバーが分からなければ [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) がデータを送信出来ないのでどうやって取得しようか考えているところです。

##### [Forward Protocol](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1)

実は、main Embulk から copy Embulk へのデータ送信のために [Forward Protocol](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1) を利用しています。これは Embulk の実装が [MessagePack](http://msgpack.org/) との親和性が高いことと [Forward Protocol](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1) を使用したサーバー・クライアント実装の開発が活発に行われていることに依ります。

現実装では以下のライブラリを使用しています。

###### [okumin/influent](https://github.com/okumin/influent)
- [Java で Fluentd の in_forward っぽいことをするライブラリを作りました #fluentd](http://qiita.com/okumin/items/ba844e3e8e6e5dc663fd)
- [Influent ベンチマーク - Part 1 #fluentd](http://blog.okumin.com/entry/2017/01/24/002953)
- [Influent ベンチマーク - Part 2 #fluentd](http://blog.okumin.com/entry/2017/04/02/233543)
- [Influent ベンチマーク - Part 3 #fluentd](http://blog.okumin.com/entry/2017/05/31/002022)

###### [komamitsu/fluency](https://github.com/komamitsu/fluency)
- [FluencyというYet another fluent Java loggerを作った話](http://komamitsu.hatenablog.com/entry/2015/12/07/003335)

# 実装上ハマったこと（Embulkガチ勢向け）
なお、書いた時点の Embulk のバージョンは v0.8.23 になります。

### [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java)
[embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) の内部で copy Embulk を実行するために利用しているクラスが [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java) になります。このクラスは Embulk の実行をコントロールしているクラスで `embulk run` や `emburk guess`、 `embulk preview` は JRuby からこのクラスを呼んで実行されています。

~~このクラスが非常に厄介で~~コンストラクタが package-local にしか公開されていないので当然 [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) の内部からは呼べません。また、このクラスのインスタンスを生成するための [Builder クラス(`Bootstrap`)](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java#L39-L118) は JRuby から呼び出されること（というか依存関係の解決を JRuby 内で行うこと）を想定しており、 `public` に公開されているメソッドを普通に使って [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java) をビルドすると依存関係が解決されていない状態の [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java) が作られてしまいます。

そこで、なんとか依存関係が解決された状態の [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java) を生成したく考えた結果が[以下のような黒魔術](https://github.com/civitaspo/embulk-filter-copy/blob/v0.0.1/src/main/java/org/embulk/filter/copy/service/EmbulkExecutorService.java#L77-L88)でした。

```EmbulkExecutorService.java
    private EmbulkEmbed newEmbulkEmbed()
    {
        try {
            Constructor<EmbulkEmbed> constructor = EmbulkEmbed.class
                    .getDeclaredConstructor(ConfigSource.class, LifeCycleInjector.class);
            constructor.setAccessible(true);
            return constructor.newInstance(null, Exec.getInjector());
        }
        catch (InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
            throw new ConfigException(e);
        }
    }
```

`Exec.getInjector()`というメソッドで依存関係が詰め込まれた`Injector`を取得出来ます。この`Injector`を無理矢理 [EmbulkEmbed](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java) のコンストラクタに詰め込んで生成するというものです。 [Builder クラス(`Bootstrap`)](https://github.com/embulk/embulk/blob/v0.8.23/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java#L39-L118) は JRuby で解決された依存関係を使って`Injector`を生成することを目的としており、この`Injector`は Embulk の実行中 Global に一つしか存在しない認識なので、使い回して問題ないだろうということでこのような実装にしてみました。他に良い方法があれば教えていただきたいです。

### static 変数
はじめはスモールにネットワーク経由でデータ送信せずにプロセス内でデータ共有すればよいかなぁと `static` な変数に `Queue` を作ってデータ共有しようと考えていました。しかしながら、Embulk は Plugin 毎に ClassLoader が異なるため、 Plugin を超えて `static` な変数の共有が出来ないのでした。（実はコードを全て読み切ったわけではないのでこの説明であってるか不安）

# 雑
- fluentd へ送信する plugin は別途外出ししても良いかもしれない
<blockquote class="twitter-tweet" data-lang="en-gb"><p lang="ja" dir="ltr">Fluentdへのforward機能欲しい <a href="https://twitter.com/hashtag/embulk?src=hash">#embulk</a></p>&mdash; Hiroshi Toyama (@toyama0919) <a href="https://twitter.com/toyama0919/status/864427688211079172">16 May 2017</a></blockquote>
<script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

# おわりに

以上になります。
それでは [embulk-filter-copy](https://github.com/civitaspo/embulk-filter-copy) をよろしくお願いいたします。