Cloud DataflowのMinimal WordCountで仕組みを理解する

以下のサイトの意訳およびメモです。
https://cloud.google.com/dataflow/examples/wordcount-example

まず基本的な機能だけ実装したMinimalWordCountを見てみる。
MinimalWordCountではGoogle Cloud Storage(GCS)からファイルを読み込み、トークン化し、ワードの数をかぞえ、GCSへファイルを出力する、シンプルなpipelineを実行する。
このコードではインプットとアウトプット先がハードコーディングされており、エラーチェックなどもしていない。Dataflow pipelineを作成するための骨組のみを記載している。
このPipelineではShakespeareの書籍でよく使われるwordをカウントする。

# まずはMinimalWordCountを動かしてみる

ソースの中でプロジェクトやバケットを指定している箇所を修正(CHANGEと書いているので見ればすぐ分かる)して、まずは実行してみる。

ソースコード
https://github.com/GoogleCloudPlatform/DataflowJavaSDK-examples/blob/master/src/main/java/com/google/cloud/dataflow/examples/MinimalWordCount.java

```
$ mvn compile exec:java  -Dexec.mainClass=com.google.cloud.dataflow.examples.MinimalWordCountDexec.mainClass=com.google.cloud.dataflow.examples.MinimalWordCount
```

CloudStorage上に集計結果のファイル(-00000-of-00006、 -0000１-of-00006...)が出力されており、ファイルの中身は以下のようになっている。

```
Fishermen: 2
Pay: 7
Exempted: 1
....
```

このプログラムでは以下の流れに沿って処理している。
1. Reading data from text files
2. Specifying 'inline' transforms
3. Counting a PCollection
4. Writing data to Cloud Storage as text files

# 1. テキストファイルからデータを読み込む
DataflowPipelineOptionsオブジェクトを生成
DataflowPipelineOptionsオブジェクトはPipelineの実行オプション（GCPのプロジェクトやバケットの場所など）をセットするためのオブジェクト。以下ではオブジェクトを作ってRunner、Project,StagingLocationを設定している。
RunnerにBlockingDataflowPipelineRunner.classを指定しているBlockingDataflowPipelineRunnerは、ジョブが起動して終了するまで待ちあわせる動きをする。

```
DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
options.setRunner(BlockingDataflowPipelineRunner.class);
options.setProject("SET_YOUR_PROJECT_ID_HERE");    options.setStagingLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_STAGING_DIRECTORY");
```

上記で作ったoptionオブジェクトを付加してPipelineオブジェクトを生成する。

```
Pipeline p = Pipeline.create(options);
```

作ったpipelineオブジェクトに対して、TextIO.Read tansformを適用する。TextIO.Readはfromで指定したファイルを読み込み、PCollectionを返す。PCollectionについては以下のページで説明されている。
https://cloud.google.com/dataflow/model/pcollection

```
p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
```

ここまででテキストファイルからデータを読み込むことができた。読み込んだデータはPCollectionオブジェクト、次のステップでこのオブジェクトを変換することにより、データ変換していく。
※PCollectionはSparkのRDDに似ている気がする。

# 2. 変換処理を行う
上記のPCollectionに対してParDoを適用する。ParDoは行の内容を個々のWordにトークン化するDoFnを呼び出す。ParDoは各要素に対して各WordのPCollection<String>を返す。
ParDoについては以下のページで説明されている。
https://cloud.google.com/dataflow/model/par-do
https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/ParDo]https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/DoFn.ProcessContext

ParDo.namedでParD PTransformを名前を付けて生成。そして.of(com.google.cloud.dataflow.sdk.transforms.DoFn<InputT, OutputT>)でDoFNを呼び出し、InputとOutputのタイプをバインドする。以下の例だと、Input,OutputともにString。
processElementをオーバライドし、c.elemntで入力値を取り出し分解。1行複数単語から1行1単語にバラしている。

```
.apply(ParDo.named("ExtractWords").of(new DoFn<String, String>() {
   @Override 
   public void processElement(ProcessContext c) {
   for (String word : c.element().split("[^a-zA-Z']+")) {
      if (!word.isEmpty()) {
         c.output(word);
         }
      }
   }
}))
```

個々のwordのPCollectionにCount transformを適用する。Count transformは新しいKey/valueペアのPCollectionを返す。ここでKeyはテキスト上のユニークな値を示している。関連付けられたvalueはwordの件数を示している。
※mapの動作をし、単語の出現回数をカウントし、Key,valueで返す。
https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/Count.PerElement

```
.apply(Count.<String>perElement())
```

Word CountのPCollectionを出力可能なString形式に変換するMapElements transformを適用する。outputファイルに出力しやすくするため。
SimpleFunction<InputT, OutputT>はPTransformを返す。 PTransformはPCollectionを入力として受取り、各要素に対してfn.apply(v)を含むPCollectionを返す。
https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/transforms/MapElements
```
.apply("FormatResults", MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
   @Override
   public String apply(KV<String, Long> input) {
      return input.getKey() + ": " + input.getValue();
   }
}))
```

Pipelineの最後に、write transformであるTextIO.Writeを適用する。TextIO.WriteはPCollectionの内容をGCSに一連のテキストファイルとして出力する。

```
.apply(TextIO.Write.to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));
```

### Pipelineを実行する。

```
p.run();
```

-------------------------------------------------------------------
importするライブラリとコメントを除いて、必要最低限フローだけに絞るとソースは以下のとおり。

```
public class MinimalWordCount {

  public static void main(String[] args) {
    DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
    options.setRunner(BlockingDataflowPipelineRunner.class);
    options.setProject("SET_YOUR_PROJECT_ID_HERE");
    options.setStagingLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_STAGING_DIRECTORY");

    Pipeline p = Pipeline.create(options);

    p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
     .apply(ParDo.named("ExtractWords").of(new DoFn<String, String>() {
                       @Override
                       public void processElement(ProcessContext c) {
                         for (String word : c.element().split("[^a-zA-Z']+")) {
                           if (!word.isEmpty()) {
                             c.output(word);
                           }
                         }
                       }
                     }))
     .apply(Count.<String>perElement())
     .apply("FormatResults", MapElements.via(new SimpleFunction<KV<String, Long>, String>() {
                       @Override
                       public String apply(KV<String, Long> input) {
                         return input.getKey() + ": " + input.getValue();
                       }
                     }))
     .apply(TextIO.Write.to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));

    p.run();
  }
}
```