embulk-output-mysqlのソースを読んでみる - 今日もプログラミング

MySQLはWebサービスでは広く使われていると思うけど、業務システムではOracleやSQL Serverも多く使われている。

しかし、今のところOracleやSQL Server用のプラグインは無いようだ。

自作するとしたらきっとMySQL用のプラグインが参考になるはずだ。

という訳で、embulk-output-mysqlのソースを読んでみることにした。

embulk-output-mysqlとembulk-output-jdbc

embulk-output-mysql プラグインはembulk-output-jdbc プラグインに依存している。

embulk-output-mysqlの各クラスは、embulk-output-jdbcのクラスを継承している。

とは言え、embulk-output-jdbcは基底クラスとしての役割だけではなく、これ自体が汎用的なDB出力用のプラグインとして使える設計のようだ。これについてもいずれ試してみたい。

MySQLOutputPlugin

このクラスでは基本的な設定を行うようだ。

以下はMySQLOutputPluginクラスの抜粋。

    @Override
    protected MySQLOutputConnector getConnector(PluginTask task, boolean retryableMetadataOperation)
    {
        MySQLPluginTask t = (MySQLPluginTask) task;

        String url = String.format("jdbc:mysql://%s:%d/%s",
                t.getHost(), t.getPort(), t.getDatabase());

        Properties props = new Properties();
        props.setProperty("user", t.getUser());
        props.setProperty("password", t.getPassword());

        props.setProperty("rewriteBatchedStatements", "true");
        props.setProperty("useCompression", "true");
        ...

接続時に使用するMySQL固有のプロパティをいろいろ設定している。

特に、rewriteBatchedStatements = trueに設定することにより大量のINSERTが高速化するらしい。

（参考：SH2の日記 MySQL Connector/Jにおける大量INSERTのチューニング）

MySQLOutputConnector

このクラスではデータベースへの接続を行っている。

しかし、DriverManager.getConnectionを使わずに直接

    this.driver = new com.mysql.jdbc.Driver();

のようにしている。

なぜだろう？

試しにDriverManager.getConnectionに変えて実行してみたら、No suitable driver foundエラーになってしまった。

通常、JARファイルのMETA-INF/services/java.sql.Driverに書いてあるJDBCドライバは自動的にロードされる。

プラグインのJARファイルは後から別のクラスローダで読み込まれるので、この機構が働かないのかな？

Javaのソースを追ってみると、

java.sql.DriverManagerのstaticイニシャライザ
↓
java.sql.DriverManager#loadInitialDrivers()
↓ 
java.util.ServiceLoader#load(Driver.class)
↓ 
java.util.ServiceLoader.LazyIterator
↓ 
ClassLoader.getSystemResources("META-INF/services/java.sql.Driver")
↓ 
getSystemClassLoader().getResources("META-INF/services/java.sql.Driver")

のように呼ぶようだ。

システムクラスローダが使われるので、やはり、別のクラスローダから読み込まれるプラグインのJARファイルは対象外ということか。

うーん、自前でMETA-INF/services/java.sql.Driverを読み込めば、JDBCドライバの自動ロードができるかもしれないな。

そうすれば、個別のDriverをnewする必要はなくなるし、embulk-output-jdbc プラグインでdriver_classを指定する必要もなくなる、と思った。

MySQLOutputConnection

このクラスは、CREATE TABLE文やINSERT文を生成するためのクラスのようだ。

テーブルのメタデータからSQLを自動生成しているが、DBMS毎の違いをこのクラスで吸収するのだろう。

    @Override
    protected String convertTypeName(String typeName)
    {
        switch(typeName) {
        case "CLOB":
            return "TEXT";
        default:
            return typeName;
        }
    }

上のようなメソッドがあって、"CLOB"を"TEXT"に変換している。

Embulkの設定ファイル（.yml）に書かれた型名（"string"など）は、プラグイン本体の中で型を表すクラス（org.embulk.spi.type.StringTypeなど）に変換される。

それが更にembulk-output-jdbcのAbstractJdbcOutputPluginクラスでデータベース型にマップされている（"CLOB"など）。

MySQLOutputConnectionクラスではそれを更に、MySQL固有の型名に変換している、ということか。

なお、embulk-output-jdbcは出力先のテーブルが存在しないときに自動生成するが、このメソッドはそのときに呼ばれるものだ。

つまり、文字列の出力先として自動生成されるときはTEXT型ということだが、あらかじめ出力先テーブルを用意しておくときはTEXT型でなくても（VARCHAR型とかでも）大丈夫だ。

MySQLBatchInsert

このクラスでは、INSERTする値の変換を行っているようだ。

スーパークラスを見ると、setBigDecimal、setInt、setString、…のように各型に対応するメソッドが定義されている。

MySQLBatchInsertクラスでは以下のメソッドをオーバーライドしている。

    @Override
    public void setFloat(float v) throws IOException, SQLException
    {
        if (Float.isNaN(v) || Float.isInfinite(v)) {
            setNull(Types.REAL);  // TODO get through argument
        } else {
            super.setFloat(v);
        }
    }

    @Override
    public void setDouble(double v) throws IOException, SQLException
    {
        if (Double.isNaN(v) || Double.isInfinite(v)) {
            setNull(Types.DOUBLE);  // TODO get through argument
        } else {
            super.setDouble(v);
        }
    }

MySQLはNaNや±∞をサポートしていないため、今のところはNULLに変換しているようだ。

ちなみに、調べたところではOracleのBINARY_FLOATやBINARY_DOUBLEはNaNや±∞をサポートしているが、SQL Serverはサポートしていない。

SQL ServerではMySQLと同じことをする必要がありそうだ。

まとめ

DBMS共通処理はembulk-output-jdbcにまとめられているので、新しいDBMS用のプラグインを作るときにオーバーライドすべき個所は多くない。

気になるのは、DBMS固有の型にどこまで対応できるか、ということかな。

これもいずれ調べてみたい。

Java	8u31
Embulk	0.4.10
embulk-output-jdbc	0.2.0
embulk-output-mysql	0.2.0