巷で知ったEmbulkプラグイン開発ノウハウ(Java編)


ちなみに当人、Javaは普段書きませんので、簡単なことでも教えていただけると非常にありがたいです。

右側の目次でやりたいことを探してください。

## カラム

### カラムを絞る方法

元ネタ

* [PageBuilderにデータを追加するときに、カラムを名前で指定して書けるようになりました](https://twitter.com/frsyuki/status/614249647075233792) v0.6.13から
* [only-columns.java](https://gist.github.com/frsyuki/54bd9a99cd0d7a2465cd)


```java
List<String> columnNames = task.getColumns();
ImmutableList.Builder<Column> builder = ImmutableList.builder();
int i = 0;
for (String columnName : columnNames) {
    for (Column inputColumn: inputSchema.getColumns()) {
        if (inputColumn.getName().equals(columnName)) {
            Column outputColumn = new Column(i++, columnName, inputColumn.getType());
            builder.add(outputColumn);
            break;
        }
    }
}
Schema outputSchema = new Schema(builder.build());
```

## 時刻

### 時刻パースの仕方(Embulk 0.6.14以降)

sonotsさんが詳しく書いてくださいました。

[Embulk の Timestamp クラスまわりのメモ (Java)](http://qiita.com/sonots/items/a422a3b0bc45fb65f017)

TimestampParser.Task をPluginTaskに追記します。

```java
public interface PluginTask
        extends Task, LineDecoder.DecoderTask, TimestampParser.Task
```

クラスのインスタンス化

```java
final TimestampParser time_parser = new TimestampParser(task.getJRuby(),"%d/%b/%Y:%T %z",task.getDefaultTimeZone());
```

パース

```java
try {
    pageBuilder.setTimestamp(3,time_parser.parse(accessLogEntryMatcher.group(4)));
} catch(TimestampParseException ex) {
  throw Throwables.propagate(ex);
}
```

[実装例](https://github.com/hiroyuki-sato/embulk-parser-apache-log/blob/master/src/main/java/org/embulk/parser/ApacheLogParserPlugin.java#L104)



### 時刻出力時のフォーマット指定

元ネタ

* [embulk の java プラグインから、Time.now.strftime 相当のことをやる方法](https://twitter.com/sonots/status/616929967822077952)

sonotsさんが詳しく書いてくださいました。

[Embulk の Timestamp クラスまわりのメモ (Java)](http://qiita.com/sonots/items/a422a3b0bc45fb65f017)

## テスト

### gemインストールせずに開発中のプラグインをテストする

[embulk のプラグイン開発時にお試し実行するには](http://qiita.com/sonots/items/4263b9c1cd07c8e25b37)

### travisでテスト

[embulk のプラグインを travis でテストする](http://qiita.com/sonots/items/bcb6902320325917a3dd)

## LocalFile

*  [LocalFileクラスの使い方1](https://twitter.com/oreradio/status/616842768472629248) [その2](https://twitter.com/oreradio/status/616842831433351168) v0.6.16 ~

## gradle 

### javaのオプション追記方法(Deprecation確認)

build.gradleに下記のエントリを追加します。

```
//
// for deprecation check
//
allprojects {
    tasks.withType(JavaCompile) {
        options.compilerArgs << "-Xlint:deprecation"
    }
}
```

### タスクの依存関係を表示する。

``gradlew tasks --all``とするとタスクの依存関係が出力されます。``--all``を省略するとトップレベルのタスクだけ出力されます。

```
/gradlew tasks --all
:tasks

------------------------------------------------------------
All tasks runnable from root project
------------------------------------------------------------

Build tasks
-----------
assemble - Assembles the outputs of this project. [jar]
build - Assembles and tests this project. [assemble, check]
buildDependents - Assembles and tests this project and all projects that depend on it.                                                                          [build]
buildNeeded - Assembles and tests this project and all projects it depends on. [build]
classes - Assembles classes 'main'.
    compileJava - Compiles Java source 'main:java'.
    processResources - Processes JVM resources 'main:resources'.
...
```

### プロジェクトの依存関係を出力

```
./gradlew dependencies
:dependencies

------------------------------------------------------------
Root project
------------------------------------------------------------

archives - Configuration for archive artifacts.
No dependencies

checkstyle - The Checkstyle libraries to be used for this project.
\--- com.puppycrawl.tools:checkstyle:6.7
     +--- org.apache.commons:commons-lang3:3.4
     +--- antlr:antlr:2.7.7
     +--- org.antlr:antlr4-runtime:4.5
     +--- commons-beanutils:commons-beanutils-core:1.8.3
     +--- commons-cli:commons-cli:1.3
     \--- com.google.guava:guava:18.0

compile - Compile classpath for source set 'main'.
+--- org.embulk:embulk-core:0.7.0
|    +--- com.google.guava:guava:18.0
|    +--- com.google.inject:guice:4.0
|    |    +--- javax.inject:javax.inject:1
|    |    +--- aopalliance:aopalliance:1.0
|    |    \--- com.google.guava:guava:16.0.1 -> 18.0
|    +--- com.google.inject.extensions:guice-multibindings:4.0
|    |    \--- com.google.inject:guice:4.0 (*)
|    +--- javax.inject:javax.inject:1
```


## 一時ファイル

### 一時ファイルの作り方

<blockquote class="twitter-tweet" lang="ja"><p lang="ja" dir="ltr">一時ファイルは、Exec.getTempFileSpace() APIを使って作成できます。将来的には、一時ファイルを保存するディスクを複数指定できるようにして、そこからround-robindで選択するような機構が追加される予定です。</p>&mdash; FURUHASHI Sadayuki (@frsyuki) <a href="https://twitter.com/frsyuki/status/616478633754923008">2015, 7月 2</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

## Embulkのソースコード整形

[Embulkのソースコードスタイルチェック方法 (Java編)](http://qiita.com/hiroysato/items/584eafb4a94a961fa571)を参照

## DynamicPageBuilder


[String→long/timestamp…の型変換を自動でやってくれて、ColumnVisitorが不要になるDynamicPageBuilderという実験的実装
](https://twitter.com/frsyuki/status/631542245255045120)

## EmbulkEmbed

* [EmbulkをJavaから実行する方法](http://www.ne.jp/asahi/hishidama/home/tech/embulk/embed.html)

## ソースの変更を検出して自動でコンパイルしてくれるgradleオプション

```
./gradlew gem -t 
```

gradleがフォアグランドで起動します。コンパイルが終わってもコマンドは終了せず次のようなメッセ〜ジが表示されます。

```
Waiting for changes to input files of tasks... (ctrl-d to exit)
```

ソースを修正すると変更を自動的に検出し次のようなメッセージを出力してコンパイルを継続します。

```
Change detected, executing build...
```

## embulk migrateの注意点

``build.gradle``に``org.embulk:embulk-core:0.7.+ ``のように``+``を利用している場合に``embulk migrate``を実行すると最新のバージョン番号に置き換わります。(Emblk 0.8.0から)

これは仕様です。[参考](https://github.com/embulk/embulk/issues/363)

build.gradleの変更前と変更後の値は次のようになります。

```diff
diff --git a/build.gradle b/build.gradle
index c3985ed..478f544 100644
--- a/build.gradle
+++ b/build.gradle
@@ -21,13 +21,13 @@ sourceCompatibility = 1.7
 targetCompatibility = 1.7
 
 dependencies {
-    compile  "org.embulk:embulk-core:0.7.+"
-    provided "org.embulk:embulk-core:0.7.+"
+    compile  "org.embulk:embulk-core:0.8.1"
+    provided "org.embulk:embulk-core:0.8.1"
     // compile "YOUR_JAR_DEPENDENCY_GROUP:YOUR_JAR_DEPENDENCY_MODULE:YOUR_JAR_DEPENDENCY_VERSION"
     compile 'org.apache.hadoop:hadoop-client:2.6.0'
     testCompile "junit:junit:4.+"
-    testCompile "org.embulk:embulk-core:0.7.+:tests"
-    testCompile "org.embulk:embulk-standards:0.7.+"
+    testCompile "org.embulk:embulk-core:0.8.1:tests"
+    testCompile "org.embulk:embulk-standards:0.8.1"
 }
```


## 最後に一度だけ実行

<blockquote class="twitter-tweet" lang="ja"><p lang="ja" dir="ltr"><a href="https://twitter.com/toyama0919">@toyama0919</a> 処理の内容にもよると思いますがcleanupの中で実行するとか。ESの例のPRだとcleanupの中でaliasの貼り替えするようにしました。 <a href="https://t.co/ugVrOLz7f6">https://t.co/ugVrOLz7f6</a></p>&mdash; oreradio (@oreradio) <a href="https://twitter.com/oreradio/status/690453896649707520">2016, 1月 22</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>


## SimpleDateFormatは使わないほうが良いらしい。

[ImmutableでスレッドセーフになったJavaの新しい日時APIの基礎知識 (1/5)](http://www.atmarkit.co.jp/ait/articles/1412/16/news041.html)

SimpleDateFormatはスレッドセーフでないらしく、誤ったデータを返すことがあるようだ。Joda-Timeを使うのが良い模様。
(パーサーは良いかもしれない(自信なし)が、フィルタでは使わないほうがよさそう)

## JavaでGuessを実装する

[embulk-parser-grok](https://github.com/arielnetworks/embulk-parser-grok/blob/master/src/main/java/org/embulk/parser/grok/GrokGuessPlugin.java)を参考にしましょう。

## Mapreduce-executor.


<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr"><a href="https://twitter.com/sonots">@sonots</a> transactionはMapReduceを発行するサーバで実行され、各タスクだけがMRで分散実行されますよ</p>&mdash; Sadayuki Furuhashi (@frsyuki) <a href="https://twitter.com/frsyuki/status/772689060141813760">2016年9月5日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr">mapreduce executorちょっと触ってみて思うのは、temp fileの書き出しが上手く動くかどうかかなあ。あそこがちゃんと動くなら問題無さそうだけど。</p>&mdash; joker1007 (きのこ派) (@joker1007) <a href="https://twitter.com/joker1007/status/772681130898100224">2016年9月5日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>


## リトライ

* [Embulkプラグインのリトライ/例外処理の実装について](http://qiita.com/sakama/items/4c5b08a50f6e2e313afa)

追補

<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr"><a href="https://twitter.com/Civitaspo">@Civitaspo</a> 書いたのが昔なので載ってないですが、接続先のAPIによってはステータス：429だけは拾ってここでtrue返してリトライした方が良さげ。<a href="https://t.co/B67zYcxud5">https://t.co/B67zYcxud5</a></p>&mdash; oreradio (@oreradio) <a href="https://twitter.com/oreradio/status/866608778560524288">2017年5月22日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr">EmbulkのRetryExecutor、例外処理+リトライの時は同じコード書かなくて便利だと思う。</p>&mdash; oreradio (@oreradio) <a href="https://twitter.com/oreradio/status/866609869670895616">2017年5月22日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr">ただjob/処理のステータスのポーリングみたいに単純にexponential back offが欲しい、例外処理は他でやってるからいらないといった場合はちょっと重すぎるかも。 <a href="https://t.co/TOAt71B0xI">https://t.co/TOAt71B0xI</a></p>&mdash; oreradio (@oreradio) <a href="https://twitter.com/oreradio/status/866609987878924288">2017年5月22日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>

## trunsationとrunでのパラメータの共有

<blockquote class="twitter-tweet" data-lang="ja"><p lang="ja" dir="ltr">embulkのtransaction -&gt; openのデータの受け渡し下記で実現できた。<br><br>- transaction<br>val t = task.dump()<br>t.set(&quot;abc&quot;, &quot;efg&quot;)<br><br>- open<br>ts.get(classOf[String] , &quot;abc&quot;)</p>&mdash; masa (@smdmts) <a href="https://twitter.com/smdmts/status/898364629813927937">2017年8月18日</a></blockquote> <script async src="//platform.twitter.com/widgets.js" charset="utf-8"></script>


## その他

* [transactionとopenを呼ぶのは別インスタンス](https://twitter.com/sonots/status/762489470054576128)
* [static変数には値を入れてはいけない](https://twitter.com/sonots/status/762491099491414016)