Dataflowのテンプレートを作ってみる

GMOアドマーケティングのT.Kです。
既存のバッチ処理をDataprocからDataflowへ移植する際にDataflowのテンプレートが気になったので、
入門ガイドのクラシック テンプレートの作成を参考にして、WordCountのコードを改修してテンプレートを作ってみます。

• テンプレート作成コマンド
  
  テンプレート作成

  Shell

  gradle clean execute -DmainClass=org.apache.beam.examples.WordCount \ -Dexec.args="--project=<YOUR_PROJECT_ID> \ --runner=DataflowRunner \ --region=us-central1 --stagingLocation=gs://<YOUR_BUCKET_NAME>/staging \ --templateLocation=gs://<YOUR_BUCKET_NAME>/templates/WordCount"

  1 2 3 4 5 6

  gradle clean execute -DmainClass=org.apache.beam.examples.WordCount \     -Dexec.args="--project=<YOUR_PROJECT_ID> \     --runner=DataflowRunner \     --region=us-central1     --stagingLocation=gs://<YOUR_BUCKET_NAME>/staging \     --templateLocation=gs://<YOUR_BUCKET_NAME>/templates/WordCount"

• コード修正なしでテンプレートを作ってみる
  IllegalArgumentExceptionが発生しました。
  
  Exception in thread "main" java.lang.IllegalArgumentException: Missing required value for [--output, "Path of the file to write to"].

  1	Exception in thread "main" java.lang.IllegalArgumentException: Missing required value for [--output, "Path of the file to write to"].

• ValueProvider対応
  パイプラインオプションを受け取る為にはValueProviderを使う必要があるので、パラメータの型をString型からValueProvider<String> に変更します。
  
  ValueProviderに変更

  Java

  public interface WordCountOptions extends PipelineOptions { /** * By default, this example reads from a public dataset containing the text of King Lear. Set * this option to choose a different input file or glob. */ @Description("Path of the file to read from") @Default.String("gs://apache-beam-samples/shakespeare/kinglear.txt") ValueProvider<String> getInputFile(); void setInputFile(ValueProvider<String> value); /** Set this required option to specify where to write the output. */ @Description("Path of the file to write to") @Required ValueProvider<String> getOutput(); void setOutput(ValueProvider<String> value); }

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19

  public interface WordCountOptions extends PipelineOptions {       /**      * By default, this example reads from a public dataset containing the text of King Lear. Set      * this option to choose a different input file or glob.      */     @Description("Path of the file to read from")     @Default.String("gs://apache-beam-samples/shakespeare/kinglear.txt")     ValueProvider<String> getInputFile();       void setInputFile(ValueProvider<String> value);       /** Set this required option to specify where to write the output. */     @Description("Path of the file to write to")     @Required     ValueProvider<String> getOutput();       void setOutput(ValueProvider<String> value);   }

• 作成は出来たがUnsupportedOperationExceptionが発生しました
  
  情報: Template successfully created. Exception in thread "main" java.lang.UnsupportedOperationException: The result of template creation should not be used.

  1 2	情報: Template successfully created. Exception in thread "main" java.lang.UnsupportedOperationException: The result of template creation should not be used.

• 原因はwaitUntilFinish
  テンプレートからDataflowRunnerを実行する場合は、waitUntilFinishがサポートされていないようです。
  waitUntilFinishを使わないように変更します。
  
  waitUntilFinish削除

  Java

  // p.run().waitUntilFinish(); p.run();

  1 2	// p.run().waitUntilFinish(); p.run();

• テンプレートを作成できました
  
  情報: Template successfully created. BUILD SUCCESSFUL in 15s

  1 2 3

  情報: Template successfully created.   BUILD SUCCESSFUL in 15s

• テンプレートの実行
  gcloudコマンドから問題なく実行できました。
  
  dataflow実行

  Shell

  gcloud dataflow jobs run <JOB_NAME> \ --region=us-central1 \ --gcs-location gs://<YOUR_BUCKET_NAME>/templates/WordCount \ --parameters inputFile=gs://apache-beam-samples/shakespeare/kinglear.txt,output=gs://<YOUR_BUCKET_NAME>/output/count

  1 2 3 4

  gcloud dataflow jobs run <JOB_NAME> \     --region=us-central1 \     --gcs-location gs://<YOUR_BUCKET_NAME>/templates/WordCount \     --parameters inputFile=gs://apache-beam-samples/shakespeare/kinglear.txt,output=gs://<YOUR_BUCKET_NAME>/output/count

• まとめ
  テンプレートにする事自体は軽い改修で対応出来ました。
  ジョブの終了待ちオプションが見当たらないので、ポーリング監視を追加する必要がありそうです。
  また、同一ソースでDirectRunnerに対応させる方法が分からなかったので、テンプレート対応は見送りました。
  引き続き調査を継続していきたいと思います。

T.K