Cloud Dataflow での Flexible Resource Scheduling の使用

このページでは、Cloud Dataflow で自動スケーリング対象のバッチ パイプラインに対して Flexible Resource Scheduling（FlexRS）を有効にする方法を説明します。

FlexRS

FlexRS は、高度なスケジューリング手法、Cloud Dataflow Shuffle サービス、プリエンプティブ仮想マシン（VM）インスタンスと通常の VM の組み合わせを使用することで、バッチ処理コストを削減します。プリエンプティブ VM と通常の VM を並列実行することで、システム イベント中に Compute Engine がプリエンプティブ VM インスタンスを停止した場合の Cloud Dataflow のユーザー エクスペリエンスが向上します。FlexRS は、プリエンプティブ VM が Compute Engine によってプリエンプトされたときに、パイプラインの処理を続行して以前の作業を確実に保持するのに役立ちます。

FlexRS を使用するジョブでは、参加とグループ化のためにサービスベースの Cloud Dataflow Shuffle を使用します。その結果、FlexRS ジョブでは一時的な計算結果を保存するために Persistent Disk リソースを使用しません。Cloud Dataflow Shuffle を使用すると、Cloud Dataflow サービスで残りのワーカーにデータを再分配する必要がないため、FlexRS ではワーカー VM のプリエンプションの処理が向上します。各 Cloud Dataflow ワーカーでは、マシンイメージと一時ログを保存するために、さらに 25 GB の Persistent Disk ボリュームが必要です。

FlexRS ジョブではスケジューリングが遅延します。そのため、FlexRS は、特定の時間枠内に完了できる毎日行われるジョブや毎週行われるジョブなど、時間の制約が厳しくないワークロードに最適です。

遅延したスケジューリング

FlexRS ジョブを送信すると、Cloud Dataflow サービスでそのジョブがキューに入れられ、ジョブの作成から 6 時間以内に実行のために送信されます。Cloud Dataflow では、利用可能な容量やその他の要因に基づいて、時間枠内のジョブを開始するのに最適な時刻が決定されます。

FlexRS ジョブを送信すると、Cloud Dataflow サービスで次の手順を実行します。

  1. ジョブの送信直後にジョブ ID を返します。
  2. 早期検証を実行します。

  3. 早期検証の結果を使用して、次のステップを決定します。

    1. 成功すると、遅延している開始を待機するためにジョブをキューに入れます。
    2. それ以外の場合、ジョブは失敗し、Cloud Dataflow サービスでエラーが報告されます。

そのため、FlexRS ジョブを送信した後、ジョブが成功すると Cloud Dataflow モニタリング インターフェースにキューに格納済みの ID とステータスが表示され、それ以外の場合は [失敗] と表示されます。

早期検証

FlexRS ジョブは送信後すぐには開始されません。早期検証中に、Cloud Dataflow サービスでは、実行パラメータと、Cloud IAM の役割やネットワーク構成などの Google Cloud Platform（GCP）環境設定を確認します。Cloud Dataflow によって、ジョブの送信時に可能な限りジョブが検証され、潜在的なエラーが報告されます。この早期検証プロセスに対しては請求されません。

早期検証ステップでは、ユーザーコードは実行されません。Apache Beam Direct Runner または FlexRS 以外のジョブを使用して問題を確認するには、コードを検証する必要があります。ジョブの作成からそのジョブの遅延されたスケジューリングまでの間に GCP 環境の変更がある場合、そのジョブは検証の初期段階で成功する可能性がありますが、それでも開始時に失敗する可能性があります。

FlexRS を有効にする

FlexRS ジョブを作成すると、そのジョブのステータスがキューに格納済みであっても、ジョブの割り当てが取得されます。したがって、FlexRS を有効にする前に、ジョブを開始するために十分な GCP プロジェクト リソースの割り当てがあることを確認してください。パブリック IP パラメータをオフにしない限り、割り当てにプリエンプティブ CPU、通常の CPU、IP アドレスの追加割り当てが含まれます。

十分な割り当てがない場合は、FlexRS ジョブのデプロイ時にアカウントに十分なリソースがない可能性があります。Cloud Dataflow では、デフォルトでワーカープール内の 90% のワーカーにプリエンプティブ VM が選択されます。CPU 割り当てを計画するときは、十分なプリエンプティブ VM 割り当てがあることを確認してください。デフォルトでは、プリエンプティブ VM 専用の割り当てはありません。明示的にプリエンプティブ VM 割り当てをリクエストする必要があります。そうしないと、FlexRS ジョブを遅延なく実行するためのリソースが不足します。

料金

FlexRS ジョブでは、次のリソースに対して請求されます。

  • 通常の CPU とプリエンプティブ CPU
  • メモリリソース
  • Cloud Dataflow Shuffle リソース
  • ワーカーあたり 25 GB の Persistent Disk リソース

Cloud Dataflow ではプリエンプティブ ワーカーと通常のワーカーの両方を使用して FlexRS ジョブを実行しますが、ワーカーのタイプにかかわらず料金は均一であり、通常の Cloud Dataflow の料金と比べると割安です。Cloud Dataflow Shuffle と Persistent Disk のリソースの料金は割引されません。

詳細については、Cloud Dataflow 料金の詳細ページをご覧ください。

FlexRS の要件

FlexRS には以下の機能が必要です。

  • Apache Beam SDK for Java 2.12.0 以上、または Apache Beam SDK for Python 2.12.0 以上。
  • Cloud Dataflow Shuffle。FlexRS をオンにすると、Cloud Dataflow Shuffle も自動的に有効になります。

注: 期間限定で、Apache Beam SDK for Java 2.5.0 以上、Apache Beam SDK for Python 2.5.0 以上、または Cloud Dataflow SDK 2.5.0 で FlexRS を有効にできます。これらの SDK に対する FlexRS のサポートは、2019 年 8 月 1 日に終了する予定です。

パイプライン オプション

Apache Beam SDK 2.12.0 以上

Java

FlexRS ジョブを有効にするには、パイプライン オプション --flexRSGoal=COST_OPTIMIZED を使用します。コスト最適化の目標は、Cloud Dataflow サービスによって、利用可能な割引リソースを選択することを意味します。

FlexRS ジョブは以下の実行パラメータに影響します。

  • numWorkers は初期ワーカー数のみを設定します。ただし、コスト管理のために maxNumWorkers を設定できます。
  • autoscalingAlgorithm=NONE を設定することはできません。
  • FlexRS ジョブには zone フラグを指定できません。Cloud Dataflow サービスでは、region パラメータで指定したリージョン内のすべての FlexRS ジョブに対してゾーンを選択します。
  • ベータ版では、デフォルトの us-central1 を使用するか、region として europe-west1 リージョン エンドポイントを指定する必要があります。
  • ベータ版では、デフォルトの n1-standard-2 を使用するか、または workerMachineType に対して n1-highmem-16 を選択する必要があります。

次の例は、FlexRS を使用するために通常のパイプライン パラメータにパラメータを追加する方法を示しています。

--flexRSGoal=COST_OPTIMIZED \
--region=europe-west1 \
--maxNumWorkers=10 \
--workerMachineType=n1-highmem-16

region、maxNumWorkers、workerMachineType を省略した場合、Cloud Dataflow サービスによりデフォルト値が決定されます。

Python

FlexRS ジョブを有効にするには、パイプライン オプション --flexrs_goal=COST_OPTIMIZED を使用します。コスト最適化の目標は、Cloud Dataflow サービスによって、利用可能な割引リソースを選択することを意味します。

FlexRS ジョブは以下の実行パラメータに影響します。

  • num_workers は初期ワーカー数のみを設定します。ただし、コスト管理のために max_num_workers を設定できます。
  • autoscaling_algorithm=NONE を設定することはできません。
  • FlexRS ジョブには zone フラグを指定できません。Cloud Dataflow サービスでは、region パラメータで指定したリージョン内のすべての FlexRS ジョブに対してゾーンを選択します。
  • ベータ版では、デフォルトの us-central1 を使用するか、region として europe-west1 リージョン エンドポイントを指定する必要があります。
  • ベータ版では、デフォルトの n1-standard-2 を使用するか、または machine_type に対して n1-highmem-16 を選択する必要があります。

次の例は、FlexRS を使用するために通常のパイプライン パラメータにパラメータを追加する方法を示しています。

--flexrs_goal=COST_OPTIMIZED \
--region=europe-west1 \
--max_num_workers=10 \
--machine_type=n1-highmem-16

region、max_num_workers、machine_type を省略した場合、Cloud Dataflow サービスによりデフォルト値が決定されます。

Apache Beam SDK 2.5.0 以下、または Cloud Dataflow SDK 2.5.0

Java

注: 期間限定で、FlexRS を Apache Beam SDK for Java 2.5.0 以上、または Cloud Dataflow SDK 2.5.0 とともに使用できます。これらの SDK に対する FlexRS のサポートは、2019 年 8 月 1 日に終了する予定です。

flexRSGoal パラメータを指定しないでください。Apache Beam SDK for Java 2.5.0 以上、または Cloud Dataflow SDK 2.5.0 とともに FlexRS ジョブを一時的に有効にするには、次のパイプライン オプションを使用します。

--experiments=flexible_resource_scheduling \
--experiments=shuffle_mode=service

flexible_resource_scheduling または shuffle_mode=service パラメータのスペルを間違えた場合でも、Cloud Dataflow サービスではエラー メッセージが報告されません。

FlexRS ジョブは以下の実行パラメータに影響します。

  • numWorkers は初期ワーカー数のみを設定します。ただし、コスト管理のために maxNumWorkers を設定できます。
  • autoscalingAlgorithm=NONE を設定することはできません。
  • FlexRS ジョブには zone フラグを指定できません。Cloud Dataflow サービスでは、region パラメータで指定したリージョン内のすべての FlexRS ジョブに対してゾーンを選択します。
  • ベータ版では、デフォルトの us-central1 を使用するか、region として europe-west1 リージョン エンドポイントを指定する必要があります。
  • ベータ版では、デフォルトの n1-standard-2 を使用するか、または workerMachineType に対して n1-highmem-16 を選択する必要があります。

次の例は、FlexRS を使用するために通常のパイプライン パラメータにパラメータを追加する方法を示しています。

--experiments=flexible_resource_scheduling \
--experiments=shuffle_mode=service \
--region=europe-west1 \
--maxNumWorkers=10 \
--workerMachineType=n1-highmem-16

region、maxNumWorkers、workerMachineType を省略した場合、Cloud Dataflow サービスによりデフォルト値が決定されます。

Python

注: 期間限定で、FlexRS を Apache Beam SDK for Python 2.5.0 以上、または Cloud Dataflow SDK 2.5.0 とともに使用できます。これらの SDK に対する FlexRS のサポートは、2019 年 8 月 1 日に終了する予定です。

flexrs_goal パラメータを指定しないでください。Apache Beam SDK for Python 2.5.0 以上、または Cloud Dataflow SDK 2.5.0 とともに FlexRS ジョブを一時的に有効にするには、次のパイプライン オプションを使用します。

--experiments=flexible_resource_scheduling \
--experiments=shuffle_mode=service

flexible_resource_scheduling または shuffle_mode=service パラメータのスペルを間違えた場合でも、Cloud Dataflow サービスではエラー メッセージが報告されません。

FlexRS ジョブは以下の実行パラメータに影響します。

  • num_workers は初期ワーカー数のみを設定します。ただし、コスト管理のために max_num_workers を設定できます。
  • autoscaling_algorithm=NONE を設定することはできません。
  • FlexRS ジョブには zone フラグを指定できません。Cloud Dataflow サービスでは、region パラメータで指定したリージョン内のすべての FlexRS ジョブに対してゾーンを選択します。
  • ベータ版では、デフォルトの us-central1 を使用するか、region として europe-west1 リージョン エンドポイントを指定する必要があります。
  • ベータ版では、デフォルトの n1-standard-2 を使用するか、または machine_type に対して n1-highmem-16 を選択する必要があります。

次の例は、FlexRS を使用するために通常のパイプライン パラメータにパラメータを追加する方法を示しています。

--experiments=flexible_resource_scheduling \
--experiments=shuffle_mode=service \
--region=europe-west1 \
--max_num_workers=10 \
--machine_type=n1-highmem-16

region、max_num_workers、machine_type を省略した場合、Cloud Dataflow サービスによりデフォルト値が決定されます。

FlexRS ジョブのモニタリング

Google Cloud Platform Console の 2 つの場所で、FlexRS ジョブのステータスをモニタリングできます。

  1. すべてのジョブが表示される [ジョブ] ページ。
  2. 送信したジョブの [ジョブの概要] ページ。

[ジョブ] ページで、開始されていないジョブのステータスは [キューに格納済み] になります。

実行中、失敗、成功の各ステータスのジョブを示す GCP Console 内の Cloud Dataflow ジョブのリスト。
図 1: 実行中、キューに格納済み、失敗、完了の各ステータスのジョブを示す GCP Console 内の Cloud Dataflow ジョブのリスト。

[ジョブの概要] ページに、キュー内で待機しているジョブに対して「ジョブがキューに追加されました。まもなく実行されます。」というメッセージが表示されます。

Cloud Dataflow モニタリング インターフェースに表示された、キューに格納済みの個々のパイプライン ジョブ。
図 2: Cloud Dataflow モニタリング インターフェースに表示された、キューに格納済みの個々のパイプライン ジョブ。

このページは役立ちましたか？評価をお願いいたします。

フィードバックを送信...

このページ
ドキュメントのフィードバック
Cloud Dataflow
サービスのフィードバック
ご不明な点がありましたら、Google のサポートページをご覧ください。