GoogleCloudで汎用Database構築2 - DataFlow2 -

GYAOのtsです。
我々のチームは、オールパブリッククラウドで、[Microservice Architecture](http://microservices.io/patterns/microservices.html)を採用した次期バックエンドを設計中です。

# 経緯
[前回の投稿](http://qiita.com/tstakano-yj/items/0ccbbde10cf177cad03b)で、DataFlowの前提知識をいくつかまとめたので、早速ハンズオンしてみる。

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

 1. pubsubにrest形式でメッセージをpublish
 1. dataflowのstream modeで処理を行う
    1. メッセージをsubscribe
    1. ランダムな文字列を発行
    1. 発行した文字列をキーにCloudDataStoreに登録。
 
pubsubにrest形式でメッセージをpublishする前に、[Apigee](http://qiita.com/tstakano-yj/items/2a6c672990333795fcda)を置いて連携したいが、それは後ほど。

# 準備
## Cloud Storage 
Cloud Storageに転送→同時にGCEのインスタンスが起動→deployのような形になっているようなので、転送先のバケットを作成する。（Eclipseからも作成可能。）
<img width="266" alt="スクリーンショット 2016-10-29 23.06.16.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/f2809da2-66c1-6411-514c-10714c897c2b.png">
↓
<img width="304" alt="スクリーンショット 2016-10-29 23.10.43.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/94d97074-3996-b392-034a-d7458bca7ea2.png">
↓
<img width="524" alt="スクリーンショット 2016-10-29 23.16.15.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/b5d45a2d-4aa5-ff4b-ba74-a68c5ed701e7.png">
flow-stagingという名前で作成。

## Cloud Pub/Sub
Cloud Pub/SubからStreamデータを購読してその内容をstoreするので、topicを作成する。
<img width="301" alt="スクリーンショット 2016-10-29 23.19.26.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/0a16e177-2ecc-445b-df9b-2aa27f7c650c.png">
↓
<img width="423" alt="スクリーンショット 2016-10-29 23.22.26.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/f3293f78-2179-6962-5190-00d1de236887.png">
↓
<img width="551" alt="スクリーンショット 2016-10-29 23.24.33.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/932f2ad1-ab1b-bee0-07e0-11572091d057.png">
テストトピックを作成。

## Cloud DataStore
特に準備の必要はなし。

# setup
ここからは[オフィシャル](https://cloud.google.com/dataflow/docs/quickstarts/quickstart-java-eclipse)に沿ってすすめる。
## Cloud SDK 
<img width="783" alt="スクリーンショット 2016-10-30 0.07.40.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/66e3733b-4d2b-ee17-b816-7a0bbb6f9ca4.png">
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の設定 [こちらのオフィシャル](https://cloud.google.com/sdk/gcloud/reference/beta/auth/application-default/login)を参照。
 - 5、6もその通りにEclipse Luna以上、JDK1.7以上の構成で。 

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

## Pipeline作成
Eclipseプラグインのお陰で簡単にプロジェクトが作成できる。（[Mavenでも構築可能](https://cloud.google.com/dataflow/docs/quickstarts/quickstart-java-maven)）
<img width="527" alt="スクリーンショット 2016-10-30 0.29.13.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/f14a2750-12cc-6132-8d6c-dc5173957541.png">
<img width="468" alt="スクリーンショット 2016-10-30 0.33.36.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/985e393b-493a-332c-f30d-d8de80997d7e.png">
すすめるとProjectIDとLocationをの入力画面があるので、入力する。

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

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

```java: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は下記の通り

```java: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で下記を作成
![スクリーンショット 2016-10-31 15.01.19.png](https://qiita-image-store.s3.amazonaws.com/0/114329/1f8e49de-9135-1f0f-a8bb-e52e5a7a57c6.png)
MainClassを指定。
![スクリーンショット 2016-10-31 15.04.05.png](https://qiita-image-store.s3.amazonaws.com/0/114329/386399ba-f81d-bd6a-1e1f-3648c76ac4fa.png)
projectIdと
staging Locationを指定。
DataflowPipelineRunnerを使用する。PipelineRunnerに関しては[前回の投稿](http://qiita.com/tstakano-yj/items/0ccbbde10cf177cad03b)を参照
Run。以上。
<img width="1138" alt="スクリーンショット 2016-10-31 15.18.03.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/283eda2d-6fe0-b273-5dbf-3f248a42cc48.png">
CloudStorageの指定バケットにStagingとして保存される。
<img width="369" alt="スクリーンショット 2016-10-31 15.19.45.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/412503c7-b47b-1d12-749c-c2fcfa567064.png">
Dataflowのコンソールでjobを確認することができる。
<img width="424" alt="スクリーンショット 2016-10-31 15.21.07.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/5869674c-5fb3-a3ab-9ee4-af79c93764a8.png">

# 実行
<img width="575" alt="スクリーンショット 2016-10-31 15.22.16.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/d97acf94-3945-b4f9-0aaf-d14784ff7a7b.png">

pubsubのtopicを見ると、subscriptionが登録されているのがわかる。
それでは実際にpubsubにpushしてみる。


<img width="1110" alt="スクリーンショット 2016-10-31 15.23.33.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/c2e8f095-37a5-c297-f2ae-633b6392e394.png">
公開ボタンから



<img width="454" alt="スクリーンショット 2016-10-31 15.25.13.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/1e6cdaca-77eb-f1b3-e21e-a2b136a089c4.png">
適当にメッセージング



<img width="1011" alt="スクリーンショット 2016-10-31 15.26.05.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/61d006bf-bd7c-1a8d-8d1e-3c7aee01d864.png">
Dataflowの画面では要素数がリアルタイムで1にかわる。



<img width="397" alt="スクリーンショット 2016-10-31 15.28.26.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/3bb4b25c-75d2-02d8-f41c-aad0776812d2.png">
Logs→ワーカーログでログを見てみると、Entityが作成されたのがわかる。



<img width="535" alt="スクリーンショット 2016-10-31 15.30.09.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/2cb0acf7-9a89-2828-885b-c03bb2a6ca9f.png">
データストアのdefault名前空間を見てみると、data kindに



<img width="727" alt="スクリーンショット 2016-10-31 15.32.16.png" src="https://qiita-image-store.s3.amazonaws.com/0/114329/91248c23-9087-94cd-2abb-6cf719e2c2e5.png">
保存を確認。

# 次回
[前回の投稿](http://qiita.com/tstakano-yj/items/0ccbbde10cf177cad03b)で、Cloud Pub/Subで順序性が保証されないことに気付いたので、設計を少し変えようかと。