embulk v0.10 APIを利用する私的メモ

はじめに

Embulkのプラグインを0.9から0.10対応にするためにのメモ。
基本的にはここJava plugins to catch up with Embulk v0.10 from v0.9に書いてあることを読めば良い。普段Javaを書かない自分用のメモ(お気づきの点があればぜひ教えてください)
@dmikurubeさんに教えてもらったものをまとめただけともいう

ゴール

• プラグインが "embulk-core" に依存しなくなること(本体のjacksonのライブラリとぶつかったり等の問題を回避して、プラグインで自由にライブラリを利用できるようにする等)

新しいAPIで書かれているプラグイン

• embulk-input-s3
• embulk-input-mixpanel
• 組み込みcsvparser

gradle-embulk-plugins

v0.10のAPIを使ってプラグインを開発するには、gradleのプラグインとして提供されるgradle-embulk-pluginsを使う必要がある、こちらのプラグインはgradleのバージョン6を必要とするため、まずgradleのバージョンをあげなければならない。

gradleのアップデート方法

gradleのバージョンは6.Xを利用する。近々gradle 7がリリースされるようだが、gradle-embulk-pluginsはまだ7.0に対応していないので、6.X系のバージョンを利用する。
最新のバージョンはここ でチェックする

./gradlew wrapper --gradle-version 7.6.1

のように実行する。

build.gradleの書き換え

How to migrate old-style build.gradle of your Embulk plugins を読みながらプラグインのbuild.gradleを書き換えする。書いてあるとおりにしたがっていけばよい。

4. Add required testCompile if depending on embulk-core:0.9.22+.の部分について、embulk-0.10.10以降は次のように記載する。

    testCompile "org.embulk:embulk-deps:0.10.26"

以下の部分は将来プラグインをrubygemsではなくmaven centralから取得できる仕組みで利用する。プラグインをMaven Centralにアップロードできるようになるまではコメントにしておくのがよい(と思う)

//publishing {
//    publications {
//        embulkPluginMaven(MavenPublication) {  // Publish it with "publishEmbulkPluginMavenPublicationToMavenRepository".
//            from components.java  // Must be "components.java". The dependency modification works only for it.
//        }
//    }
//    repositories {
//        maven {
//            url = "${project.buildDir}/mavenPublishLocal"
//        }
//    }
//}

Javaコンパイルの際にdeprecatedの警告を出力するには以下の設定を追記する。

tasks.withType(JavaCompile) {
    options.compilerArgs << "-Xlint:deprecation" << "-Xlint:unchecked"
}

gradle-embulk-pluginsはプラグインロードに必要なRubyファイルを自動生成するようになっており、従来のプラグインにあったプラグイン名.rbというファイルは不要になっている。
諸般の事情(guessをRubyで実装している等)がある場合には以下のような記述をして独自のRubyファイルをgemに含めることができる。

gem {

    generateRubyCode = false  // Avoiding generate lib/embulk/parser/jsonpath.rb
    into("lib/embulk/parser/") {
      from "lib/embulk/parser/jsonpath.rb"
    }
    into("lib/embulk/guess/") {
      from "lib/embulk/guess/jsonpath.rb"
    }

Javaの書き換え

importの書き換え

従来embulk-coreの中にまとまっていたクラス群は、embulk-util-configやembulk-util-timestampなど別のライブラリに分割された。import文をそれぞれに置き換えなければならない。

以下のようなスクリプトを作成して、機械的に変換をおこなっている。

#!/bin/bash

set -ue

fatal(){
    echo "Fatal Error: $*" >&2
    exit 1
}

if [ ! -d src ] ; then
  fatal "invalid exec directory"
fi

find src -name '*.java' -print0 | xargs -0 \
  perl -i -pe 's/import org.embulk.config.Config;/import org.embulk.util.config.Config;/;
               s/import org.embulk.config.ConfigDefault;/import org.embulk.util.config.ConfigDefault;/;
               s/import org.embulk.config.Task;/import org.embulk.util.config.Task;/;
               s/import org.embulk.spi.ColumnConfig;/import org.embulk.util.config.units.ColumnConfig;/;
               s/import org.embulk.spi.SchemaConfig;/import org.embulk.util.config.units.SchemaConfig;/;
               s/import com.google.common.base.Optional;/import java.util.Optional;/;
               s/import org.embulk.spi.util.FileInputInputStream;/import org.embulk.util.file.FileInputInputStream;/;
               s/import org.embulk.spi.time.TimestampParser;/import org.embulk.util.timestamp.TimestampFormatter;/;
'

Exec.getModelManager()

今後なくなっていく予定だが、当面はExecInternalを利用することで暫定的に回避することができる。ModelManagerは近々削除される予定になっており、いずれ別の書き方に変える必要がある。

Exec.getModelManager()

ExecInternal.getModelManager

ExecInternalを利用するには、build.gradleにcompileOnly "embulk-core"する必要があるが、それ自体非推奨になっている。つまりできるだけembulk-coreをdependenciesに入れるのは避けるべきである。

テストでは、ExecInternalはしばらく利用できる(v0.11開発時に検討される模様)
以下の記述は、テストだけでExecInternalを利用するようにするための設定

dependencies {
    testCompile "org.embulk:embulk-core:0.10.28"
}    

loadConfig, loadTask

ConfigMapper,TaskMapperそれぞれのクラスで提供されるmapメソッドを利用する。

@@ -43,6 +46,7 @@ public class JsonpathParserPlugin
 {
 
     private static final Logger logger = LoggerFactory.getLogger(JsonpathParserPlugin.class);
+    private static final ConfigMapperFactory CONFIG_MAPPER_FACTORY = ConfigMapperFactory.builder().addDefaultModules().build();
 
     private static final Configuration JSON_PATH_CONFIG = Configuration
             .builder()
@@ -94,18 +98,22 @@ public class JsonpathParserPlugin
     @Override
     public void transaction(ConfigSource config, ParserPlugin.Control control)
     {
-        PluginTask task = config.loadConfig(PluginTask.class);
+        final ConfigMapper configMapper = CONFIG_MAPPER_FACTORY.createConfigMapper();
+        final PluginTask task = configMapper.map(config, PluginTask.class);
 
         Schema schema = getSchemaConfig(task).toSchema();
 
-        control.run(task.dump(), schema);
+        control.run(task.toTaskSource(), schema); // v0.10 only
     }
 
     @Override
     public void run(TaskSource taskSource, Schema schema,
             FileInput input, PageOutput output)
     {
-        PluginTask task = taskSource.loadTask(PluginTask.class);
+        final TaskMapper taskMapper = CONFIG_MAPPER_FACTORY.createTaskMapper();
+        final PluginTask task = taskMapper.map(taskSource, PluginTask.class);
+

依存ライブラリの固定

./gradlew dependencies --write-locks を使って利用しているバージョンの依存関係を固定する

ロックファイルのあるべき姿

build.gradleが適切に設定されていれば、以下の行が含まれているはず

javax.validation:validation-api:1.1.0.Final
com.fasterxml.jackson.core:jackson-annotations:2.6.7
com.fasterxml.jackson.core:jackson-core:2.6.7
com.fasterxml.jackson.core:jackson-databind:2.6.7
com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.6.7

理想的ではない設定

依存関係が不適切だと、例えば、以下のようになることがある。

embulk-util-configはcom.fasterxml.jackson.core:jackson-databindの2.6.7を利用してようとしているが、
別のライブラリの依存関係の影響で2.6.7の代わりに2.6.7.2が使われることになっている。

+--- org.embulk:embulk-util-aws-credentials:0.4.0
|    +--- org.embulk:embulk-util-config:0.1.1 -> 0.2.1
|    |    +--- javax.validation:validation-api:1.1.0.Final
|    |    +--- com.fasterxml.jackson.core:jackson-annotations:2.6.7
|    |    +--- com.fasterxml.jackson.core:jackson-core:2.6.7
|    |    +--- com.fasterxml.jackson.core:jackson-databind:2.6.7 -> 2.6.7.2 (*)
|    |    \--- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.6.7
|    |         +--- com.fasterxml.jackson.core:jackson-core:2.6.7
|    |         \--- com.fasterxml.jackson.core:jackson-databind:2.6.7 -> 2.6.7.2 (*)

このような出力がされる場合にはロックファイルには、以下のように2.6.7ではなく2.6.7.2が記載される。

com.fasterxml.jackson.core:jackson-databind:2.6.7.2

このようになる場合は適切にexcludeを利用して依存関係からパッケージを除外して、上記のあるべき姿になるように設定をする

build.gradleの記述の基本方針

• プラグイン内で、jacksonに依存していないことが明らかであれば、embulk-util-configでexcludeを入れる必要はない
• embulk-util-configでexcludeをした場合は、明示的にcompileを記述して、上記のパッケージ全部が入るようにすること

以下は@dmikurubeさんによる補足

Jackson のバージョン

下記 以外 の Jackson 系のライブラリが embulkPluginRuntime.lockfile に入り込んだ場合は (Embulk v0.9.23 でも動かすには) それも 2.6.7 に固定しなければならない。

(Embulk v0.9 系 v0.10.31 以下をあきらめて v0.10.32 (未リリース) 以上や v0.11 (未リリース) のみを対象にするなら 2.6.7 以外でもいい。 2.6.7 以外の Jackson が入っていても Embulk v0.9 系で動くこともあるが、保証はない。運。)

com.fasterxml.jackson.core:jackson-annotations
com.fasterxml.jackson.core:jackson-core
com.fasterxml.jackson.core:jackson-databind
com.fasterxml.jackson.datatype:jackson-datatype-jdk8

Jackson 以外で関連するバージョン

以下が embulkPluginRuntime.lockfile に入り込んだ場合は、どのライブラリから派生してこれらが入り込んだのかを突き止め、 exclude などを駆使して以下のとおりにバージョンを固定しなければならない。

(Embulk v0.9 系 v0.10.31 以下をあきらめて v0.10.32 (未リリース) 以上や v0.11 (未リリース) のみを対象にするなら、これ以外のバージョンでもいい。これ以外のバージョンが入っていても Embulk v0.9 系で動くこともあるが、保証はない。運。)

• com.google.guava:guava => 18.0
• org.apache.commons:commons-lang3 => 3.4
• joda-time:joda-time => 2.9.2
• org.apache.bval:bval-jsr303 => 0.5

常に exclude して、入れ直してはならないもの

以下が embulkPluginRuntime.lockfile に入り込んだ場合は exclude して、入らないようにしなければならない。

• org.slf4j:slf4j-api
• org.msgpack:msgpack-core

バージョン固定のために exclude した場合

上記 slf4j-api と msgpack-core 以外を、バージョン固定のために exclude した場合は、その指定バージョンを compile "..." で入れ直すのを忘れてはならない。

忘れると、新しい Embulk バージョンでは動かない。 (そのライブラリに必要な依存関係がどこにもないことになるので)

補足ここまで

API互換

ドキュメントのsinceのところが0.10と書かれていたら新しいAPI、v0.9を利用する場合このメソッドは使ってはいけない

TypeModule

次のようなエラーが出た場合、ConfigMapperFactory を作るときに TypeModule を add する必要がある。(embulk-util-config:0.3.0でデフォルトで、TypeModuleも追加するようになったのでこの修正は不要参考)

Caused by: java.lang.IllegalArgumentException: 
　　Can not construct instance of org.embulk.spi.type.Type,
  　problem: abstract types either need to be mapped to concrete types,

    private static final ConfigMapperFactory CONFIG_MAPPER_FACTORY = ConfigMapperFactory
            .builder()
            .addDefaultModules()
            .addModule(new TypeModule())　// <-- これ
            .build();

参考

脱Guava

ImmutableMapの書き換え

diff --git a/src/main/java/org/embulk/parser/jsonpath/JsonpathParserPlugin.java b/src/main/java/org/embulk/parser/jsonpath/JsonpathParserPlugin.java
index 1946175..5bc9746 100644
--- a/src/main/java/org/embulk/parser/jsonpath/JsonpathParserPlugin.java
+++ b/src/main/java/org/embulk/parser/jsonpath/JsonpathParserPlugin.java
@@ -2,8 +2,10 @@ package org.embulk.parser.jsonpath;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.JsonNodeType;
+
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.Optional;
-import com.google.common.collect.ImmutableMap;
 import com.jayway.jsonpath.Configuration;
 import com.jayway.jsonpath.InvalidJsonException;
 import com.jayway.jsonpath.JsonPath;
@@ -186,15 +188,15 @@ public class JsonpathParserPlugin
 
     private Map<Column, String> createJsonPathMap(PluginTask task, Schema schema)
     {
-        ImmutableMap.Builder<Column, String> builder = ImmutableMap.builder();
+        Map<Column, String> columnMap = new HashMap<>();
         for (int i = 0; i < schema.size(); i++) {
             ColumnConfig config = getSchemaConfig(task).getColumn(i);
             JsonpathColumnOption option = config.getOption().loadConfig(JsonpathColumnOption.class);
             if (option.getPath().isPresent()) {
-                builder.put(schema.getColumn(i), option.getPath().get());
+                columnMap.put(schema.getColumn(i), option.getPath().get());
             }
         }
-        return builder.build();
+        return Collections.unmodifiableMap(columnMap);
     }
 
     private void skipOrThrow(DataException cause, boolean stopOnInvalidRecord)

ImmutableSet

diff --git a/src/main/java/org/embulk/parser/jsonpath/ColumnVisitorImpl.java b/src/main/java/org/embulk/parser/jsonpath/ColumnVisitorImpl.java
index d846ffe..d5aef46 100644
--- a/src/main/java/org/embulk/parser/jsonpath/ColumnVisitorImpl.java
+++ b/src/main/java/org/embulk/parser/jsonpath/ColumnVisitorImpl.java
@@ -3,8 +3,9 @@ package org.embulk.parser.jsonpath;
 import com.fasterxml.jackson.databind.JsonNode;
 
 import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
 im#port java.util.Optional;
-import com.google.common.collect.ImmutableList;
 import org.embulk.parser.jsonpath.JsonpathParserPlugin.PluginTask;
 import org.embulk.parser.jsonpath.JsonpathParserPlugin.TypecastColumnOption;
 import org.embulk.spi.Column;
@@ -31,8 +32,8 @@ public class ColumnVisitorImpl
         implements ColumnVisitor
 {
     private static final JsonParser JSON_PARSER = new JsonParser();
-    private static final List<String> BOOL_TRUE_STRINGS = ImmutableList.of("true", "1", "yes", "on", "y", "t");
-    private static final List<String> BOOL_FALSE_STRINGS = ImmutableList.of("false", "0", "no", "off", "n", "f");
+    private static final List<String> BOOL_TRUE_STRINGS = Collections.unmodifiableList(Arrays.asList("true", "1", "yes", "on", "y", "t"));
+    private static final List<String> BOOL_FALSE_STRINGS = Collections.unmodifiableList(Arrays.asList("false", "0", "no", "off", "n", "f"));
 
     protected final PluginTask task;
     protected final Schema schema;

Throwable

diff --git a/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java b/embulk-core
/src/main/java/org/embulk/EmbulkEmbed.java
index af0e88f4..2f98c9b2 100644
--- a/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java
+++ b/embulk-core/src/main/java/org/embulk/EmbulkEmbed.java
@@ -5,7 +5,6 @@ import static com.google.common.base.Preconditions.checkState;

 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Function;
-import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.inject.Injector;
@@ -277,7 +276,10 @@ public class EmbulkEmbed {
         try {
             injector.destroy();
         } catch (Exception ex) {
-            throw Throwables.propagate(ex);
+            if (ex instanceof RuntimeException) {
+                throw (RuntimeException) ex;
+            }
+            throw new RuntimeException(ex);
         }
     }

テスト

embulk 0.10.1 ~ 0.10.29まで、-Lと-Iのオプションの挙動が0.9と異なっていた。
embulk 0.10.30でこの挙動は修正された

ここを参照(後日追加)

0.9との互換性

v0.10 APIを作って、embulk 0.9で動かした際に以下のようなエラーが出た場合

org.embulk.exec.PartialExecutionException: java.lang.RuntimeException: java.lang.NoSuchMethodError: org.embulk.spi.Buffer: method <init>()V not found
	at org.embulk.exec.BulkLoader$LoaderState.buildPartialExecuteException(BulkLoader.java:340)
	at org.embulk.exec.BulkLoader.doRun(BulkLoader.java:566)
	at org.embulk.exec.BulkLoader.access$000(BulkLoader.java:35)
	at org.embulk.exec.BulkLoader$1.run(BulkLoader.java:353)
	at org.embulk.exec.BulkLoader$1.run(BulkLoader.java:350)
	at org.embulk.spi.Exec.doWith(Exec.java:22)
	at org.embulk.exec.BulkLoader.run(BulkLoader.java:350)
	at org.embulk.EmbulkEmbed.run(EmbulkEmbed.java:242)
	at org.embulk.EmbulkRunner.runInternal(EmbulkRunner.java:291)
	at org.embulk.EmbulkRunner.run(EmbulkRunner.java:155)
	at org.embulk.cli.EmbulkRun.runSubcommand(EmbulkRun.java:431)
	at org.embulk.cli.EmbulkRun.run(EmbulkRun.java:90)
	at org.embulk.cli.Main.main(Main.java:64)

build.gradleに以下のエントリを入れてビルドする。バージョンは0.1.3以降でなければならない。

dependencies {
    // ... snip ..
    compile "org.embulk:embulk-util-file:0.1.3"

Task

TimestampFormatter.Task

    // From org.embulk.spi.time.TimestampFormatter.Task
    @Config("default_timezone")
    @ConfigDefault("\"UTC\"")
    public String getDefaultTimeZoneId();

    // From org.embulk.spi.time.TimestampFormatter.Task
    @Config("default_timestamp_format")
    @ConfigDefault("\"%Y-%m-%d %H:%M:%S.%6N %z\"")
    public String getDefaultTimestampFormat();

LineEncoder.Task

    // From org.embulk.spi.util.LineEncoder.Task
    @Config("charset")
    @ConfigDefault("\"utf-8\"")
    public Charset getCharset();

    // From org.embulk.spi.util.LineEncoder.Task
    @Config("newline")
    @ConfigDefault("\"CRLF\"")
    public Newline getNewline();