Presto コネクターを実装する第一回

こんにちは。松本です。

弊社ではマーケティングや CRM 関連の SaaS を開発／運営していますが、近年、この領域で利用するデータの規模がますます大きくなっています。またそれに伴い、データの格納先も従来のように「RDB のみ」というわけにはいかず用途によって様々に分かれてきています。

しかしこれらのデータをシステム上で統合して扱おうとすると、その方法に頭を悩ませます。Hadoop/HDFS ベースにデータを集め、Hive 等を使ってバッチ処理させるというアプローチも良いですが、お客様やプロダクト企画チームからの期待は「よりインタラクティブ」な処理です。

Facebook 社によって公開されたオープンソースの分散処理基盤である Presto はこういった課題に対するソリューションです。

そこで今回は Presto のインストール方法について記事を書こう・・・と思ったのですが、スズキ編集長から「それだけじゃ面白くないじゃん」的な~~理不尽な~~発言を頂いたので、Presto を更に活用することを可能にする Presto コネクターの実装方法について、何回かに分けて記事を書くことにしました。

Presto コネクターとは？

コネクターは Presto が各種データソースにアクセスすることを可能にするアダプター的なもので、本記事執筆時点で Java8 ベースで Service Provider が使われています。

JAR File Specification - Service Provider
http://docs.oracle.com/javase/8/docs/technotes/guides/jar/jar.html#Service_Provider

既存のコネクター実装として、Cassandra、Hive、MySQL、PostgreSQL 等のコネクターが提供されています。

Presto Document - 4. Connectors
https://prestodb.io/docs/current/connector.html

Presto コネクター開発手順

実は現時点ではコネクター開発に関するドキュメントは少なく、オープンソースで公開されている実際のコネクターのソースを見ながら開発するしかありません。
# javadoc にも期待できません・・。

Presto Document - 9. Developer Guide
http://facebook.github.io/presto/docs/current/develop.html

Github facebook/presto
https://github.com/facebook/presto

Github にある presto-example-http というのがコネクターのサンプルとなっています。

presto-example-http
https://github.com/facebook/presto/tree/master/presto-example-http

本記事ではこの presto-example-http を使って解説してもよかったのですが、入門編としてわかりやすい記事になるよう、更にシンプルなオリジナルコネクターを作りながら解説することにしました。

開発手順はざっくりと次の通り。

Presto Plugin インターフェースの取得
コネクターの実装
プロパティファイルの作成
ファイルの配置
Presto の設定変更
Presto の再起動
動作確認

この順番で解説していきます（連載途中で変更するかもしれませんが）。

1. Presto Plugin インターフェースの取得

本記事を書いている時点で Presto 最新バージョンは 0.101 ですが、都合上今回は 0.96 を使います（私の開発環境が現在 0.96 なのです。0.101 に上げる時間が取れなかったというだけです。ごめんなさい）。

コネクター開発では presto-spi を使います。

その他、必要なものを含めた build.gradle は次のようになります。

apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'propdeps'
apply plugin: 'propdeps-maven'

sourceCompatibility = 1.8
version = '1.0'

buildscript {
    repositories {
        maven { url 'http://repo.spring.io/plugins-release' }
        mavenCentral()
    }
    dependencies {
        classpath 'org.springframework.build.gradle:propdeps-plugin:0.0.6'
    }
}

repositories {
    mavenCentral()
}

dependencies {
    provided 'org.projectlombok:lombok:1.16.2'

    provided 'com.facebook.presto:presto-spi:0.96'

    provided 'com.fasterxml.jackson.core:jackson-databind:2.5.0'
    provided 'com.fasterxml.jackson.core:jackson-core:2.5.0'
    provided 'com.fasterxml.jackson.core:jackson-annotations:2.5.0'

    provided 'com.google.guava:guava:18.0'
    provided 'com.google.inject:guice:3.0'
    provided 'com.google.inject.extensions:guice-multibindings:3.0'

    provided 'javax.validation:validation-api:1.0.0.GA'

    provided 'io.airlift:json:0.103'
    provided 'io.airlift:log:0.103'
    provided 'io.airlift:configuration:0.103'
    provided 'io.airlift:bootstrap:0.103'
    provided 'io.airlift:slice:0.10'
    compile 'io.airlift:units:0.103'
}

apply plugin: 'java'

apply plugin: 'eclipse'

apply plugin: 'propdeps'

apply plugin: 'propdeps-maven'

sourceCompatibility = 1.8

version = '1.0'

buildscript {

repositories {

maven { url 'http://repo.spring.io/plugins-release' }

mavenCentral()

}

dependencies {

classpath 'org.springframework.build.gradle:propdeps-plugin:0.0.6'

}

repositories {

mavenCentral()

}

dependencies {

provided 'org.projectlombok:lombok:1.16.2'

provided 'com.facebook.presto:presto-spi:0.96'

provided 'com.fasterxml.jackson.core:jackson-databind:2.5.0'

provided 'com.fasterxml.jackson.core:jackson-core:2.5.0'

provided 'com.fasterxml.jackson.core:jackson-annotations:2.5.0'

provided 'com.google.guava:guava:18.0'

provided 'com.google.inject:guice:3.0'

provided 'com.google.inject.extensions:guice-multibindings:3.0'

provided 'javax.validation:validation-api:1.0.0.GA'

provided 'io.airlift:json:0.103'

provided 'io.airlift:log:0.103'

provided 'io.airlift:configuration:0.103'

provided 'io.airlift:bootstrap:0.103'

provided 'io.airlift:slice:0.10'

compile 'io.airlift:units:0.103'

}

ほとんどのライブラリは Presto 本体の lib ディレクトリに置かれているため、provided コンフィギュレーションで指定しています。

lombok は必須ではありませんが、本記事を書くにあたり趣旨と関係の薄い無駄なコードを省くことが出来るので利用することにしました（lombok についてはいずれ本ブログ内で誰かが紹介記事を書くと思います）。

lombok のインストール方法は下記を参考にして下さい。

Project Lombok - Download
http://projectlombok.org/download.html

CodeZine - Java特有の冗長なコードを簡潔に記述する「Lombok」
http://codezine.jp/article/detail/7274

2. Presto コネクターの実装

クラス構成

本記事では固定のテーブル／データを扱う ReadOnly なコネクターを開発します。

最小構成で実現する為、実装クラスは次の十数ファイル程度になります。

TechscorePrestoPlugin implements com.facebook.presto.spi.Plugin

Service Provider の実装クラスです。TechscoreConnectorFactory オブジェクトを生成する役割を担います。

TechscoreConnectorFactory implements com.facebook.presto.spi.ConnectorFactory

その名の通り、コネクターのファクトリクラスで、TechscoreConnector オブジェクトを生成します。コネクターオブジェクトの生成には com.google.inject.Injector を使い、必要な情報をインジェクションします。

TechscoreModule implements com.google.inject.Module

コネクタ生成時のインジェクションで使います。

TechscoreConnector implements com.facebook.presto.spi.Connector

コネクターオブジェクトです。コネクターとして必要な各種オブジェクトを返すだけのクラスです。

TechscoreMetadata extends com.facebook.presto.spi.ReadOnlyConnectorMetadata

スキーマやテーブルに関するメタデータを返すクラスです。

TechscoreHandleResolver implements com.facebook.presto.spi.ConnectorHandleResolver

各種ハンドルクラスを扱う為のリゾルバークラスです。ハンドルクラスには以下のようなテーブル情報を扱うものやカラム情報を扱うもの等があります。

TechscoreTableHandle implements com.facebook.presto.spi.ConnectorTableHandle

テーブル情報を扱う為のクラスです。 ConnectorTableHandle 自体は空のインターフェースですので、何を実装するかは開発者に委ねられています。

TechscoreColumnHandle implements com.facebook.presto.spi.ConnectorColumnHandle

カラム情報を扱う為のクラスです。 ConnectorColumnHandle 自体は空のインターフェースですので、何を実装するかは開発者に委ねられています。
※ 0.101 では ColumnHandle に変更されたようです。

TechscoreSplitManager implements com.facebook.presto.spi.ConnectorSplitManager

コネクターがデータを効率よく取得するためのパーティショニングを担うクラスです。結果として TechscorePartition オブジェクトや、TechscoreSplit オブジェクトを生成します。

TechscorePartition implements com.facebook.presto.spi.ConnectorPartition

TechscoreSplitManager によって生成されるパーティショニング情報を持つクラスです。

TechscoreSplit implements com.facebook.presto.spi.ConnectorSplit

パーティショニングに基づき決定されたデータソースへのアクセス情報を扱うクラスです。 TechscoreRecordSet オブジェクトの生成時に使用されます。

TechscoreRecordSetProvider implements com.facebook.presto.spi.ConnectorRecordSetProvider

TechscoreRecordSet オブジェクトを生成する役割を担います。

TechscoreRecordSet implements com.facebook.presto.spi.RecordSet

TechscoreSplit オブジェクト単位で生成され、TechscoreRecordCursor オブジェクトを生成する役割を担います。

TechscoreRecordCursor implements com.facebook.presto.spi.RecordCursor

データ操作に扱う「カーソル」で、 JDBC の java.sql.ResultSet と同じような役割を担います。

TechscoreConnectorConfig

カーソル専用のプロパティファイルの情報がバインディングされる POJO です。

その他、今回のサンプルとして固定のテーブル／データを定義するクラス等も必要となりますが、標準で必要となるクラスは上記の通りとなります。

Plugin とその周辺クラスの実装

JAR の Service Provider 仕様に基づき、本コネクタにおいて Plugin インターフェースの実装クラスが TechscorePrestoPlugin であることを示す必要があります。

やり方は、下記テキストファイルを用意し、その中に実装クラス名を記載するだけです。

java/main/resources/META-INF/services/com.facebook.presto.spi.Plugin

com.facebook.presto.spi.Plugin.TechscorePrestoPlugin

1	com.facebook.presto.spi.Plugin.TechscorePrestoPlugin

TechscorePrestoPlugin のソースです。

com.techscore.example.presto.plugin.TechscorePrestoPlugin

package com.techscore.example.presto.plugin;

import java.util.List;
import java.util.Map;

import lombok.NonNull;

import com.facebook.presto.spi.ConnectorFactory;
import com.facebook.presto.spi.Plugin;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;

public class TechscorePrestoPlugin implements Plugin {

    private Map<String, String> optionalConfig = ImmutableMap.of();

    public synchronized Map<String, String> getOptionalConfig() {
        return optionalConfig;
    }

    @Override
    public synchronized void setOptionalConfig(@NonNull Map<String, String> optionalConfig) {
        this.optionalConfig = ImmutableMap.copyOf(optionalConfig);
    }

    @Override
    public synchronized <T> List<T> getServices(Class<T> type) {
        if (type == ConnectorFactory.class) {
            return ImmutableList.of(type.cast(new TechscoreConnectorFactory(getOptionalConfig())));
        }
        return ImmutableList.of();
    }
}

package com.techscore.example.presto.plugin;

import java.util.List;

import java.util.Map;

import lombok.NonNull;

import com.facebook.presto.spi.ConnectorFactory;

import com.facebook.presto.spi.Plugin;

import com.google.common.collect.ImmutableList;

import com.google.common.collect.ImmutableMap;

public class TechscorePrestoPlugin implements Plugin {

private Map<String, String> optionalConfig = ImmutableMap.of();

public synchronized Map<String, String> getOptionalConfig() {

return optionalConfig;

}

@Override

public synchronized void setOptionalConfig(@NonNull Map<String, String> optionalConfig) {

this.optionalConfig = ImmutableMap.copyOf(optionalConfig);

}

@Override

public synchronized <T> List<T> getServices(Class<T> type) {

if (type == ConnectorFactory.class) {

return ImmutableList.of(type.cast(new TechscoreConnectorFactory(getOptionalConfig())));

}

return ImmutableList.of();

}

getServices() メソッドで、ConnectorFactory 実装クラスを要求された時に TechscoreConnectorFactory オブジェクトを生成しています。この時コンストラクタに渡されるコンフィグは Presto によってプロパティファイルから読み込まれ、setOptionalConfig() メソッドを経由してセットされています。

生成される TechscoreConnectorFactory のソースは次の通り。

com.techscore.example.presto.plugin.TechscoreConnectorFactory

package com.techscore.example.presto.plugin;

import io.airlift.bootstrap.Bootstrap;
import io.airlift.json.JsonModule;

import java.util.Map;

import lombok.Getter;
import lombok.NonNull;

import com.facebook.presto.spi.Connector;
import com.facebook.presto.spi.ConnectorFactory;
import com.google.common.base.Throwables;
import com.google.inject.Binder;
import com.google.inject.Injector;
import com.google.inject.Module;
import com.google.inject.name.Names;

public class TechscoreConnectorFactory implements ConnectorFactory {

    private static final String NAME = "techscore";

    @Getter
    private final Map<String, String> optionalConfig;

    public TechscoreConnectorFactory(@NonNull Map<String, String> optionalConfig) {
        this.optionalConfig = optionalConfig;
    }

    @Override
    public String getName() {
        return NAME;
    }

    @Override
    public Connector create(@NonNull final String connectorId, @NonNull Map<String, String> config) {
        try {
            Bootstrap app = new Bootstrap(new JsonModule(),
                    new TechscoreModule(connectorId), new Module() {
                        @Override
                        public void configure(Binder binder) {
                            binder.bindConstant().annotatedWith(Names.named("connectorId")).to(connectorId);
                        }
                    });

            Injector injector = app.strictConfig()
                    .doNotInitializeLogging()
                    .setRequiredConfigurationProperties(config)
                    .setOptionalConfigurationProperties(getOptionalConfig())
                    .initialize();

            return injector.getInstance(TechscoreConnector.class);
        } catch (Exception e) {
            throw Throwables.propagate(e);
        }
    }
}

package com.techscore.example.presto.plugin;

import io.airlift.bootstrap.Bootstrap;

import io.airlift.json.JsonModule;

import java.util.Map;

import lombok.Getter;

import lombok.NonNull;

import com.facebook.presto.spi.Connector;

import com.facebook.presto.spi.ConnectorFactory;

import com.google.common.base.Throwables;

import com.google.inject.Binder;

import com.google.inject.Injector;

import com.google.inject.Module;

import com.google.inject.name.Names;

public class TechscoreConnectorFactory implements ConnectorFactory {

private static final String NAME = "techscore";

@Getter

private final Map<String, String> optionalConfig;

public TechscoreConnectorFactory(@NonNull Map<String, String> optionalConfig) {

this.optionalConfig = optionalConfig;

}

@Override

public String getName() {

return NAME;

}

@Override

public Connector create(@NonNull final String connectorId, @NonNull Map<String, String> config) {

try {

Bootstrap app = new Bootstrap(new JsonModule(),

new TechscoreModule(connectorId), new Module() {

@Override

public void configure(Binder binder) {

binder.bindConstant().annotatedWith(Names.named("connectorId")).to(connectorId);

}

});

Injector injector = app.strictConfig()

.doNotInitializeLogging()

.setRequiredConfigurationProperties(config)

.setOptionalConfigurationProperties(getOptionalConfig())

.initialize();

return injector.getInstance(TechscoreConnector.class);

} catch (Exception e) {

throw Throwables.propagate(e);

}

create() メソッドでは、com.google.inject.Injector を使って必要な情報をインジェクションしています。

ここでインジェクトする情報は TechscoreModule を使って設定しています。

com.techscore.example.presto.plugin.TechscoreModule

package com.techscore.example.presto.plugin;

import static io.airlift.configuration.ConfigurationModule.bindConfig;
import lombok.Getter;
import lombok.NonNull;

import com.google.inject.Binder;
import com.google.inject.Module;
import com.google.inject.Scopes;

public class TechscoreModule implements Module {

    @Getter
    private final String connectorId;

    public TechscoreModule(@NonNull String connectorId) {
        this.connectorId = connectorId;
    }

    @Override
    public void configure(Binder binder) {
        binder.bind(TechscoreConnector.class).in(Scopes.SINGLETON);
        binder.bind(TechscoreMetadata.class).in(Scopes.SINGLETON);
        binder.bind(TechscoreClient.class).in(Scopes.SINGLETON);
        binder.bind(TechscoreSplitManager.class).in(Scopes.SINGLETON);
        binder.bind(TechscoreRecordSetProvider.class).in(Scopes.SINGLETON);
        binder.bind(TechscoreHandleResolver.class).in(Scopes.SINGLETON);
        bindConfig(binder).to(TechscoreConnectorConfig.class);
    }
}

package com.techscore.example.presto.plugin;

import static io.airlift.configuration.ConfigurationModule.bindConfig;

import lombok.Getter;

import lombok.NonNull;

import com.google.inject.Binder;

import com.google.inject.Module;

import com.google.inject.Scopes;

public class TechscoreModule implements Module {

@Getter

private final String connectorId;

public TechscoreModule(@NonNull String connectorId) {

this.connectorId = connectorId;

}

@Override

public void configure(Binder binder) {

binder.bind(TechscoreConnector.class).in(Scopes.SINGLETON);

binder.bind(TechscoreMetadata.class).in(Scopes.SINGLETON);

binder.bind(TechscoreClient.class).in(Scopes.SINGLETON);

binder.bind(TechscoreSplitManager.class).in(Scopes.SINGLETON);

binder.bind(TechscoreRecordSetProvider.class).in(Scopes.SINGLETON);

binder.bind(TechscoreHandleResolver.class).in(Scopes.SINGLETON);

bindConfig(binder).to(TechscoreConnectorConfig.class);

}

configure() メソッドを見ると、何をインジェクション可能としているか明らかですね。TechscorePrestoPlugin に渡されたコンフィグはここで TechscoreConnectorConfig オブジェクトにマッピングされます。

com.techscore.example.presto.plugin.TechscoreConnectorConfig

package com.techscore.example.presto.plugin;

import io.airlift.configuration.Config;

import javax.validation.constraints.NotNull;

public class TechscoreConnectorConfig {

    private String urlBase;

    @NotNull
    public String getUrlBase() {
        return urlBase;
    }

    @Config("techscore.url-base")
    public TechscoreConnectorConfig setUrlBase(String urlBase) {
        this.urlBase = urlBase;
        return this;
    }
}

package com.techscore.example.presto.plugin;

import io.airlift.configuration.Config;

import javax.validation.constraints.NotNull;

public class TechscoreConnectorConfig {

private String urlBase;

@NotNull

public String getUrlBase() {

return urlBase;

}

@Config("techscore.url-base")

public TechscoreConnectorConfig setUrlBase(String urlBase) {

this.urlBase = urlBase;

return this;

}

今回のサンプルではコンフィグオブジェクトに本サイトのベース URL を持たせます。setUrlBase() メソッドのように、@Config でプロパティファイルのキーを指定することでその値がプロパティにセットされます。

プロパティファイルは次のような内容となります（配置場所は次回以降に）。

techscore.properties

connector.name=techscore
techscore.url-base=http://www.techscore.com/blog/

1 2	connector.name=techscore techscore.url-base=http://www.techscore.com/blog/

そして TechscoreConnector がこれ。

com.techscore.example.presto.plugin.TechscoreConnector

package com.techscore.example.presto.plugin;

import javax.inject.Inject;

import lombok.Getter;
import lombok.NonNull;

import com.facebook.presto.spi.Connector;
import com.facebook.presto.spi.ConnectorHandleResolver;
import com.facebook.presto.spi.ConnectorIndexResolver;
import com.facebook.presto.spi.ConnectorMetadata;
import com.facebook.presto.spi.ConnectorPageSourceProvider;
import com.facebook.presto.spi.ConnectorRecordSetProvider;
import com.facebook.presto.spi.ConnectorRecordSinkProvider;
import com.facebook.presto.spi.ConnectorSplitManager;

public class TechscoreConnector implements Connector {

    @Getter
    private final ConnectorHandleResolver handleResolver;

    @Getter
    private final ConnectorMetadata metadata;

    @Getter
    private final ConnectorSplitManager splitManager;

    @Getter
    private final ConnectorRecordSetProvider recordSetProvider;

    @Inject
    public TechscoreConnector(@NonNull TechscoreHandleResolver tHandleResolver,
            @NonNull TechscoreMetadata tMetadata,
            @NonNull TechscoreSplitManager tSplitManager,
            @NonNull TechscoreRecordSetProvider tRecordSetProvider) {
        this.handleResolver = tHandleResolver;
        this.metadata = tMetadata;
        this.splitManager = tSplitManager;
        this.recordSetProvider = tRecordSetProvider;
    }

    @Override
    public ConnectorPageSourceProvider getPageSourceProvider() {
        throw new UnsupportedOperationException();
    }

    @Override
    public ConnectorRecordSinkProvider getRecordSinkProvider() {
        throw new UnsupportedOperationException();
    }

    @Override
    public ConnectorIndexResolver getIndexResolver() {
        throw new UnsupportedOperationException();
    }
}

package com.techscore.example.presto.plugin;

import javax.inject.Inject;

import lombok.Getter;

import lombok.NonNull;

import com.facebook.presto.spi.Connector;

import com.facebook.presto.spi.ConnectorHandleResolver;

import com.facebook.presto.spi.ConnectorIndexResolver;

import com.facebook.presto.spi.ConnectorMetadata;

import com.facebook.presto.spi.ConnectorPageSourceProvider;

import com.facebook.presto.spi.ConnectorRecordSetProvider;

import com.facebook.presto.spi.ConnectorRecordSinkProvider;

import com.facebook.presto.spi.ConnectorSplitManager;

public class TechscoreConnector implements Connector {

@Getter

private final ConnectorHandleResolver handleResolver;

@Getter

private final ConnectorMetadata metadata;

@Getter

private final ConnectorSplitManager splitManager;

@Getter

private final ConnectorRecordSetProvider recordSetProvider;

@Inject

public TechscoreConnector(@NonNull TechscoreHandleResolver tHandleResolver,

@NonNull TechscoreMetadata tMetadata,

@NonNull TechscoreSplitManager tSplitManager,

@NonNull TechscoreRecordSetProvider tRecordSetProvider) {

this.handleResolver = tHandleResolver;

this.metadata = tMetadata;

this.splitManager = tSplitManager;

this.recordSetProvider = tRecordSetProvider;

}

@Override

public ConnectorPageSourceProvider getPageSourceProvider() {

throw new UnsupportedOperationException();

}

@Override

public ConnectorRecordSinkProvider getRecordSinkProvider() {

throw new UnsupportedOperationException();

}

@Override

public ConnectorIndexResolver getIndexResolver() {

throw new UnsupportedOperationException();

}

前述の通り、コンストラクタに必要なオブジェクトがインジェクションされます。また、これらのオブジェクトは lombok の @Getter アノテーションによって自動生成されたゲッターメソッドによってアクセス可能となります。

今回不要なメソッドは UnsupportedOperationException をスローするよう実装しています。

次回は

長い割には序文的な記事になってしまいましたが、ここまでで概ね準備が整いました。
次回はメタデータおよびハンドル系のクラスの実装に進みます。