GYAOのtsです。
我々のチームは、オールパブリッククラウドで、Microservice Architectureを採用した次期バックエンドを設計中です。

経緯

前回の投稿で、DataFlowの前提知識をいくつかまとめたので、早速ハンズオンしてみる。

今回の目指すところ

今回もチーム内のメンバー全員でイベントチックに構築してみる。
やりたいことは以下。

pubsubにrest形式でメッセージをpublish
dataflowのstream modeで処理を行う
1. メッセージをsubscribe
2. ランダムな文字列を発行
3. 発行した文字列をキーにCloudDataStoreに登録。

pubsubにrest形式でメッセージをpublishする前に、Apigeeを置いて連携したいが、それは後ほど。

準備

Cloud Storage

Cloud Storageに転送→同時にGCEのインスタンスが起動→deployのような形になっているようなので、転送先のバケットを作成する。（Eclipseからも作成可能。）

↓

↓

flow-stagingという名前で作成。

Cloud Pub/Sub

Cloud Pub/SubからStreamデータを購読してその内容をstoreするので、topicを作成する。

↓

↓

テストトピックを作成。

Cloud DataStore

特に準備の必要はなし。

setup

ここからはオフィシャルに沿ってすすめる。

Cloud SDK

Source: https://cloud.google.com/dataflow/docs/quickstarts/quickstart-java-eclipse

1~3は普通に作成、有効にする。
4　Eclipseで開発するマシンにCloudSDKをインストール。
- install後、下記を実行。
  - gcloud beta auth application-default login
  - eclipseからのauthの設定こちらのオフィシャルを参照。
5、6もその通りにEclipse Luna以上、JDK1.7以上の構成で。

Eclipse

今回はEclipse neonを使用。
jdk1.8
Eclipseにはアップデートサイトhttps://dl.google.com/dataflow/eclipse/ からプラグインをインストールしておく。

Pipeline作成

Eclipseプラグインのお陰で簡単にプロジェクトが作成できる。（Mavenでも構築可能）

すすめるとProjectIDとLocationをの入力画面があるので、入力する。

項目	概要
ProjectID	自分のプロジェクトのID
Location	先程作成したCloudStorageのLocation。この場合はgs://flow-staging　プルダウンで出てきます。

Coding

Pipelineは指定されたクラスのmainメソッドの中で構築する。Google提供のサンプルを少しいじって以下のようにしてみた。

StoreJob.java


    public static void main(String[] args) throws IOException {

        Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
        options.setDataset("xxxxxx"); //ProjectId
        options.setKind("data"); //Datastoreのkind
        options.setNamespace("default"); //Datastoreのnamespace
        options.setJobName("store-job"); //Deployされたあと、一覧に出るJobの名前。
        options.setStreaming(true); //streamingモードの有効化
        Pipeline pipeline = Pipeline.create(options); //パイプライン作成

        PCollection<String> input;
        LOG.info("Reading from PubSub.");
        input = pipeline.apply(PubsubIO.Read.named("subscriber").topic("projects/xxxxxxxxx/topics/test")) //Pub/Subの購読をapply
                        .apply(Window.<String>into(SlidingWindows.of(Duration.standardMinutes(1)).every(Duration.standardSeconds(10)))); //pubsubのinputはunboundedなので、boundedの処理を行うためにwindowを使用（1分間間隔のwindow。10秒毎に新しいwindowを生成）
        input.apply(ParDo.of(new CreateEntityFn(options.getNamespace(), options.getKind()))) //
                .apply(DatastoreIO.v1().write().withProjectId(options.getDataset()));
        pipeline.run();
    }

CreateEntityFnは下記の通り

CreateEntityFn.java

    static class CreateEntityFn extends DoFn<String, Entity> {

        private final String namespace;
        private final String kind;
        private final com.google.datastore.v1.Key ancestorKey;

        CreateEntityFn(String namespace, String kind) {
            this.namespace = namespace;
            this.kind = kind;
            ancestorKey = makeAncestorKey(namespace, kind);
        }

        public Entity makeEntity(String content) {
            Entity.Builder entityBuilder = Entity.newBuilder();
            Key.Builder keyBuilder = makeKey(ancestorKey, kind, UUID.randomUUID().toString());
            if (namespace != null) {
                keyBuilder.getPartitionIdBuilder().setNamespaceId(namespace);
            }
            else {
                //do nothing.
            }
            entityBuilder.setKey(keyBuilder.build());
            entityBuilder.getMutableProperties().put("content", makeValue(content).build());
            LOG.info("create Entity included content : " + content);
            return entityBuilder.build();
        }

        @Override
        public void processElement(ProcessContext c) {
            c.output(makeEntity(c.element()));
        }
    }

Deploy

EclipseのRun→Run Configurationsで下記を作成

MainClassを指定。

projectIdと
staging Locationを指定。
DataflowPipelineRunnerを使用する。PipelineRunnerに関しては前回の投稿を参照
Run。以上。

CloudStorageの指定バケットにStagingとして保存される。

Dataflowのコンソールでjobを確認することができる。