ワークフローを使用する

次のようにして、ワークフローを設定して実行します。

1. ワークフロー テンプレートを作成する
2. マネージド（エフェメラル）クラスタを構成するか、既存のクラスタを選択する
3. ジョブを追加する
4. テンプレートをインスタンス化してワークフローを実行する

ワークフロー テンプレートをパラメータ化して、さまざまなワークフローに対して動的に使用できます。また、YAML ファイルを使用するか InstantiateInline API を呼び出すことで、ワークフロー テンプレート リソースの作成も変更もしないインライン ワークフローを定義して実行することもできます。

テンプレートを作成する

gcloud CLI

次の command を実行して、Dataproc ワークフロー テンプレート リソースを作成します。

gcloud dataproc workflow-templates create TEMPLATE_ID \
    --region=REGION

メモ:

• REGION: テンプレートが実行されるリージョンを指定します。
• TEMPLATE_ID: テンプレートの ID（「workflow-template-1」など）を指定します。
• CMEK 暗号化。--kms-key フラグを追加して、ワークフロー テンプレート ジョブ引数で CMEK 暗号化を使用できます。

REST API

WorkflowTemplate を workflowTemplates.create リクエストの一部として送信します。WorkflowTemplate.EncryptionConfig.kmsKey フィールドを追加して、ワークフロー テンプレート ジョブ引数で CMEK 暗号化を使用できます。kmsKey

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

クラスタの構成または選択

Dataproc では、ワークフロー用の新しい「マネージド」クラスタを作成することも既存のクラスタを使用することもできます。

• 既存のクラスタ: ワークフローでのクラスタ セレクタの使用を確認して、ワークフローに応じて既存のクラスタを選択します。

• マネージド クラスタ: ワークフローに応じてマネージド クラスタを構成する必要があります。Dataproc により、ワークフロー ジョブを実行するための新しいクラスタが作成されます。ワークフローが終了すると、そのクラスタは削除されます。

  gcloud コマンドライン ツールまたは Dataproc API を使用すると、ワークフローのマネージド クラスタを構成できます。

  gcloud コマンド

  gcloud dataproc cluster create から継承したフラグを使用して、マネージド クラスタ（ワーカー数、マスターおよびワーカーのマシンタイプなど）を構成します。Dataproc は、一意性を確保するために、クラスタ名に接尾辞を追加します。 --service-account フラグを使用して、マネージド クラスタ用の VM サービス アカウントを指定できます。

  gcloud dataproc workflow-templates set-managed-cluster TEMPLATE_ID \
      --region=REGION \
      --master-machine-type=MACHINE_TYPE \
      --worker-machine-type=MACHINE_TYPE \
      --num-workers=NUMBER \
      --cluster-name=CLUSTER_NAME
      --service-account=SERVICE_ACCOUNT
  

  REST API

  WorkflowTemplatePlacement.ManagedCluster をご覧ください。こちらは WorkflowTemplate リクエストまたは workflowTemplates.create リクエストによって送信された、完成済みの workflowTemplates.update の一部として提供できます。

  GceClusterConfig.serviceAccount フィールドを使用して、マネージド クラスタ用の VM サービス アカウントを指定できます。

  Console

  Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

テンプレートにジョブを追加する

1 つ以上のジョブ依存関係を指定しない限り、すべてのジョブは同時に実行されます。ジョブの依存関係は、その他のジョブのリストとして表されます。これらのジョブは、最終的なジョブを開始する前に完了しておく必要があります。ジョブごとに step-id を指定する必要があります。この ID はワークフロー内で一意である必要がありますが、グローバルに一意である必要はありません。

gcloud コマンド

gcloud dataproc jobs submit から継承したジョブタイプとフラグを使用して、テンプレートに追加するジョブを定義します。必要に応じて ‑‑start-after job-id of another workflow job フラグを使用して、ワークフローにある 1 つ以上の他のジョブが完了してから、目的のジョブが開始されるようにすることもできます。

現在、--max-failures-per-hour と --max-failures-per-hour の再実行可能なジョブフラグは、Dataproc ワークフロー テンプレート ジョブではサポートされていません。

例:

Hadoop ジョブ「foo」を「my-workflow」テンプレートに追加します。

gcloud dataproc workflow-templates add-job hadoop \
    --region=REGION \
    --step-id=foo \
    --workflow-template=my-workflow \
    -- space separated job args

ジョブ「bar」を「my-workflow」テンプレートに追加します。このジョブは、ワークフロー内のジョブ「foo」が正常に完了してから実行されます。

gcloud dataproc workflow-templates add-job JOB_TYPE \
    --region=REGION \
    --step-id=bar \
    --start-after=foo \
    --workflow-template=my-workflow \
    -- space separated job args

別のジョブ「baz」を「my-workflow」テンプレートに追加します。このジョブは「foo」と「bar」の両方のジョブが正常に完了してから実行されます。

gcloud dataproc workflow-templates add-job JOB_TYPE \
    --region=REGION \
    --step-id=baz \
    --start-after=foo,bar \
    --workflow-template=my-workflow \
    -- space separated job args

REST API

WorkflowTemplate.OrderedJob をご覧ください。このフィールドは、workflowTemplates.create または workflowTemplates.update リクエストによって送信された、完成済みの WorkflowTemplate の一部として提供されます。

現在、maxFailuresPerHour と maxFailuresTotal の OrderedJob.JobScheduling フィールドは、Dataproc ワークフロー テンプレート ジョブではサポートされていません。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

現在、Max restarts per hour の再実行可能なジョブ オプションは、Dataproc ワークフロー テンプレート ジョブではサポートされていません。

ワークフローの実行

ワークフロー テンプレートをインスタンス化すると、テンプレートで定義されたワークフローが実行されます。テンプレートの複数のインスタンス化を行えます。つまり、1 つのワークフローを複数回実行できます。

gcloud コマンド

gcloud dataproc workflow-templates instantiate TEMPLATE_ID \
    --region=REGION

このコマンドにより、オペレーション ID が返されます。これを使用して、ワークフローのステータスを追跡できます。

コマンドと出力の例:

gcloud beta dataproc workflow-templates instantiate my-template-id \
    --region=us-central1
...
WorkflowTemplate [my-template-id] RUNNING
...
Created cluster: my-template-id-rg544az7mpbfa.
Job ID teragen-rg544az7mpbfa RUNNING
Job ID teragen-rg544az7mpbfa COMPLETED
Job ID terasort-rg544az7mpbfa RUNNING
Job ID terasort-rg544az7mpbfa COMPLETED
Job ID teravalidate-rg544az7mpbfa RUNNING
Job ID teravalidate-rg544az7mpbfa COMPLETED
...
Deleted cluster: my-template-id-rg544az7mpbfa.
WorkflowTemplate [my-template-id] DONE

REST API

workflowTemplates.instantiate をご覧ください。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

ワークフロー ジョブの失敗

ワークフロー内のいずれかのジョブが失敗すると、ワークフローが失敗します。 Dataproc は、障害の影響を小さくするため、同時に実行されているすべてのジョブを停止させ、以降のジョブが開始されないようにします。

ワークフローのモニタリングと一覧表示

gcloud コマンド

ワークフローをモニタリングするには、次のコマンドを実行します。

gcloud dataproc operations describe OPERATION_ID \
    --region=REGION

注: この operation-id は、gcloud dataproc workflow-templates instantiate を使用してワークフローをインスタンス化すると返されます（ワークフローの実行をご覧ください）。

ワークフローのステータスを一覧表示するには、次のコマンドを実行します。

gcloud dataproc operations list \
    --region=REGION \
    --filter="labels.goog-dataproc-operation-type=WORKFLOW AND status.state=RUNNING"

REST API

ワークフローをモニタリングするには、Dataproc の operations.get API を使用します。

実行中のワークフローを一覧表示するには、ラベルフィルタを設定して、Dataproc の operations.list API を使用します。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

ワークフローの終了

Google Cloud CLI を使用するか、Dataproc API を呼び出して、ワークフローを終了できます。

注: ワークフローを終了すると、実行中のワークフロー ジョブがキャンセルされます。また、ワークフローがマネージド（エフェメラル）クラスタで実行されている場合、マネージド クラスタが削除されます。

gcloud コマンド

gcloud dataproc operations cancel OPERATION_ID \
    --region=REGION

注: この operation-id は、gcloud dataproc workflow-templates instantiate を使用してワークフローをインスタンス化すると返されます（ワークフローの実行をご覧ください）。

REST API

operations.cancel API をご覧ください。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

ワークフロー テンプレートの更新

更新しても、実行中のワークフローには影響がありません。新しいテンプレート バージョンは、新しいワークフローにのみ適用されます。

gcloud コマンド

ワークフロー テンプレートは、既存のワークフロー template-id を参照する、次の新しい gcloud workflow-templates コマンドを発行することで更新できます。

• gcloud dataproc workflow-templates add job
• gcloud dataproc workflow-templates set-managed-cluster
• gcloud dataproc workflow-templates set-cluster-selector

上記のコマンドを既存のワークフロー テンプレートに対して発行します。

REST API

REST API でテンプレートを更新するには:

1. workflowTemplates.get を呼び出します。version フィールドに現在のサーバーのバージョンが入力された、現在のテンプレートが返されます。
2. 取得したテンプレートを更新します。
3. 更新後のテンプレートで workflowTemplates.update を呼び出します。

同時変更の防止対策として、テンプレート更新のリクエストで、現在のサーバー バージョンを workflowTemplate.version フィールドで指定する必要があります。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。

ワークフロー テンプレートの削除

gcloud コマンド

gcloud dataproc workflow-templates delete TEMPLATE_ID \
    --region=REGION

注: この operation-id は、gcloud dataproc workflow-templates instantiate を使用してワークフローをインスタンス化すると返されます（ワークフローの実行をご覧ください）。

REST API

workflowTemplates.delete をご覧ください。

Console

Google Cloud コンソールの Dataproc [ワークフロー] ページから、既存のワークフロー テンプレートとインスタンス化されたワークフローを表示できます。