Google Cloud Dataflow Runner の使用

Google Cloud Dataflow Runner は、Cloud Dataflow マネージドサービスを使用します。Cloud Dataflow サービスでパイプラインを実行すると、ランナーは実行可能なコードと依存関係を Google Cloud Storage バケットにアップロードし、Google Cloud Platform のマネージドリソースでパイプラインを実行する Cloud Dataflow ジョブを作成します。

Cloud Dataflow Runner とサービスは大規模な連続ジョブに適しており、以下の機能を提供します。

• フルマネージドサービス
• ジョブのライフタイム全体にわたるワーカー数の自動スケーリング
• 動的な作業の再調整

Beam 機能マトリックスには、Cloud Dataflow Runner でサポートされている機能が記載されています。

Cloud Dataflow Runner の前提条件と設定

Cloud Dataflow Runner を使用するには、選択した言語のCloud Dataflow クイックスタートの「始める前に」セクションの設定を完了する必要があります。

1. Google Cloud Platform Console プロジェクトを選択または作成します。
2. プロジェクトの課金を有効にします。
3. 必要な Google Cloud API（Cloud Dataflow、Compute Engine、Stackdriver Logging、Cloud Storage、Cloud Storage JSON、Cloud Resource Manager）を有効にします。パイプラインコードで BigQuery、Cloud Pub/Sub、Cloud Datastore などの API を使用する場合は、追加で有効にする必要がある場合があります。
4. Google Cloud Platform で認証します。
5. Google Cloud SDK をインストールします。
6. Cloud Storage バケットを作成します。

依存関係の指定

Java を使用する場合、`pom.xml` で Cloud Dataflow Runner への依存関係を指定する必要があります。

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
  <version>2.60.0</version>
  <scope>runtime</scope>
</dependency>

このセクションは、Python 用 Beam SDK には適用されません。

自己実行JAR

このセクションは、Python 用 Beam SDK には適用されません。

Apache AirFlowなどのスケジューラを使用してパイプラインを開始する場合など、自己完結型アプリケーションが必要になる場合があります。前のセクションに示されている既存の依存関係を追加するだけでなく、`pom.xml` のプロジェクトセクションに次の依存関係を明示的に追加することで、自己実行 JAR をパックできます。

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
    <version>${beam.version}</version>
    <scope>runtime</scope>
</dependency>

次に、Maven JAR プラグインに `mainClass` 名を追加します。

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-jar-plugin</artifactId>
  <version>${maven-jar-plugin.version}</version>
  <configuration>
    <archive>
      <manifest>
        <addClasspath>true</addClasspath>
        <classpathPrefix>lib/</classpathPrefix>
        <mainClass>YOUR_MAIN_CLASS_NAME</mainClass>
      </manifest>
    </archive>
  </configuration>
</plugin>

`mvn package -Pdataflow-runner` を実行した後、`ls target` を実行すると、（アーティファクトID が `beam-examples`、バージョンが 1.0.0 であると仮定すると）次の出力が表示されます。

beam-examples-bundled-1.0.0.jar

Cloud Dataflow で自己実行 JAR を実行するには、次のコマンドを使用します。

java -jar target/beam-examples-bundled-1.0.0.jar \
  --runner=DataflowRunner \
  --project=<YOUR_GCP_PROJECT_ID> \
  --region=<GCP_REGION> \
  --tempLocation=gs://<YOUR_GCS_BUCKET>/temp/ \
  --output=gs://<YOUR_GCS_BUCKET>/output

Cloud Dataflow Runner のパイプラインオプション

Cloud Dataflow Runner（Java）でパイプラインを実行する場合は、以下の一般的なパイプラインオプションを検討してください。 Cloud Dataflow Runner（Python）でパイプラインを実行する場合は、以下の一般的なパイプラインオプションを検討してください。

その他のパイプライン構成オプションについては、DataflowPipelineOptions `PipelineOptions` インターフェース（およびすべてのサブインターフェース）のリファレンスドキュメントを参照してください。

追加情報と注意事項

ジョブのモニタリング

パイプラインの実行中は、Dataflow モニタリングインターフェースまたはDataflow コマンドラインインターフェースを使用して、ジョブの進行状況を監視し、実行の詳細を表示し、パイプラインの結果に関する更新を受信できます。

ブロッキング実行

ジョブが完了するまでブロックするには、`pipeline.run()` から返された `PipelineResult` で `waitToFinish``wait_until_finish` を呼び出します。Cloud Dataflow Runner は、待機中にジョブステータスの更新とコンソールメッセージを出力します。結果がアクティブなジョブに接続されている間は、コマンドラインから **Ctrl + C** を押してもジョブはキャンセルされないことに注意してください。ジョブをキャンセルするには、Dataflow モニタリングインターフェースまたはDataflow コマンドラインインターフェースを使用します。

ストリーミング実行

パイプラインで非有界データソースまたはシンクを使用する場合は、`streaming` オプションを `true` に設定する必要があります。

ストリーミング実行を使用する場合は、以下の考慮事項に留意してください。

1. ストリーミングパイプラインは、ユーザーが明示的にキャンセルしない限り終了しません。ストリーミングジョブは、Dataflow モニタリングインターフェースから、またはDataflow コマンドラインインターフェース（gcloud dataflow jobs cancel コマンド）を使用してキャンセルできます。

2. ストリーミングジョブは、デフォルトで Google Compute Engine マシンプラス `n1-standard-2` 以上を使用します。`n1-standard-2` はストリーミングジョブを実行するために必要な最小マシンプラスであるため、これをオーバーライドしないでください。

3. ストリーミング実行の料金はバッチ実行とは異なります。