既存のパイプラインを更新する

gcloud

gcloud dataflow jobs update-options コマンドを使用します。

gcloud dataflow jobs update-options \
  --region=REGION \
  --min-num-workers=MINIMUM_WORKERS \
  --max-num-workers=MAXIMUM_WORKERS \
  --worker-utilization-hint=TARGET_UTILIZATION \
  JOB_ID

次のように置き換えます。

• REGION: ジョブのリージョンの ID
• MINIMUM_WORKERS: Compute Engine インスタンスの最小数
• MAXIMUM_WORKERS: Compute Engine インスタンスの最大数
• TARGET_UTILIZATION: [0.1, 0.9] の範囲内の値
• JOB_ID: 更新するジョブの ID

--min-num-workers、--max-num-workers、worker-utilization-hint を個別に更新することもできます。

REST

projects.locations.jobs.update メソッドを使用します。

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?updateMask=MASK
{
  "runtime_updatable_params": {
    "min_num_workers": MINIMUM_WORKERS,
    "max_num_workers": MAXIMUM_WORKERS,
    "worker_utilization_hint": TARGET_UTILIZATION
  }
}

次のように置き換えます。

• MASK: 更新するパラメータのカンマ区切りリスト。次のいずれかから選択します。
  • runtime_updatable_params.max_num_workers
  • runtime_updatable_params.min_num_workers
  • runtime_updatable_params.worker_utilization_hint
• PROJECT_ID: Dataflow ジョブの Google Cloud プロジェクト ID
• REGION: ジョブのリージョンの ID
• JOB_ID: 更新するジョブの ID
• MINIMUM_WORKERS: Compute Engine インスタンスの最小数
• MAXIMUM_WORKERS: Compute Engine インスタンスの最大数
• TARGET_UTILIZATION: [0.1, 0.9] の範囲内の値

min_num_workers、max_num_workers、worker_utilization_hint を個別に更新することもできます。更新するパラメータを updateMask クエリ パラメータで指定し、更新された値をリクエスト本文の runtimeUpdatableParams フィールドに含めます。次の例では、min_num_workers を更新します。

PUT https://dataflow.googleapis.com/v1b3/projects/my_project/locations/us-central1/jobs/job1?updateMask=runtime_updatable_params.min_num_workers
{
  "runtime_updatable_params": {
    "min_num_workers": 5
  }
}

Java

• --update オプションを渡します。
• PipelineOptions で --jobName オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• --dataflowServiceOptions=graph_validate_only サービス オプションを含めます。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transformNameMapping オプションで渡す必要があります。
• 新しいバージョンの Apache Beam SDK を使用する置換ジョブを送信する場合は、--updateCompatibilityVersion を元のジョブで使用されている Apache Beam SDK のバージョンに設定します。

Python

• --update オプションを渡します。
• PipelineOptions で --job_name オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• --dataflow_service_options=graph_validate_only サービス オプションを含めます。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform_name_mapping オプションで渡す必要があります。
• 新しいバージョンの Apache Beam SDK を使用する置換ジョブを送信する場合は、--updateCompatibilityVersion を元のジョブで使用されている Apache Beam SDK のバージョンに設定します。

Go

• --update オプションを渡します。
• --job_name オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• --dataflow_service_options=graph_validate_only サービス オプションを含めます。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform_name_mapping オプションで渡す必要があります。

gcloud

Flex テンプレート ジョブのジョブグラフを検証するには、additional-experiments オプションを指定して gcloud dataflow flex-template run コマンドを使用します。

• --update オプションを渡します。
• JOB_NAME は、更新するジョブと同じ名前に設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• --additional-experiments=graph_validate_only オプションを含めます。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform-name-mappings オプションで渡す必要があります。

例:

gcloud dataflow flex-template run JOB_NAME --additional-experiments=graph_validate_only

JOB_NAME は、更新するジョブの名前に置き換えます。

REST

FlexTemplateRuntimeEnvironment（Flex テンプレート）または RuntimeEnvironment オブジェクトの additionalExperiments フィールドを使用します。

{
  additionalExperiments : ["graph_validate_only"]
  ...
}

Java

• --update オプションを渡します。
• PipelineOptions で --jobName オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transformNameMapping オプションで渡す必要があります。
• 新しいバージョンの Apache Beam SDK を使用する置換ジョブを送信する場合は、--updateCompatibilityVersion を元のジョブで使用されている Apache Beam SDK のバージョンに設定します。

Python

• --update オプションを渡します。
• PipelineOptions で --job_name オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform_name_mapping オプションで渡す必要があります。
• 新しいバージョンの Apache Beam SDK を使用する置換ジョブを送信する場合は、--updateCompatibilityVersion を元のジョブで使用されている Apache Beam SDK のバージョンに設定します。

Go

• --update オプションを渡します。
• --job_name オプションに、更新するジョブと同じ名前を設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform_name_mapping オプションで渡す必要があります。

gcloud

gcloud CLI を使用して Flex テンプレート ジョブを更新するには、gcloud dataflow flex-template run コマンドを使用します。gcloud CLI を使用した他のジョブの更新はサポートされていません。

• --update オプションを渡します。
• JOB_NAME は、更新するジョブと同じ名前に設定します。
• --region オプションに、更新するジョブのリージョンと同じリージョンを設定します。
• パイプライン内の変換名が変更された場合は、変換マッピングを用意して --transform-name-mappings オプションで渡す必要があります。

REST

以下では、REST API を使用してテンプレート以外のジョブを更新する方法について説明します。REST API を使用してクラシック テンプレート ジョブを更新するには、カスタム テンプレート ストリーミング ジョブを更新するをご覧ください。REST API を使用して Flex テンプレート ジョブを更新するには、Flex テンプレート ジョブを更新するをご覧ください。

1. projects.locations.jobs.get メソッドを使用して、置換するジョブの job リソースを取得します。値 JOB_VIEW_DESCRIPTION を持つ view クエリ パラメータを含めます。JOB_VIEW_DESCRIPTION を含めると、後続のリクエストがサイズ上限を超えないように、レスポンスのデータ量が制限されます。より詳細なジョブ情報が必要な場合は、値 JOB_VIEW_ALL を使用します。

   GET https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID?view=JOB_VIEW_DESCRIPTION
   

   次の値を置き換えます。

   • PROJECT_ID: Dataflow ジョブの Google Cloud プロジェクト ID
   • REGION: 更新するジョブのリージョン
   • JOB_ID: 更新するジョブのジョブ ID

2. ジョブを更新するには、projects.locations.jobs.create メソッドを使用します。リクエストの本文で、取得した job リソースを使用します。

   POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   {
     "id": JOB_ID,
     "replaceJobId": JOB_ID,
     "name": JOB_NAME,
     "type": "JOB_TYPE_STREAMING",
     "transformNameMapping": {
       string: string,
       ...
     },
   }
   

   次のように置き換えます。

   • JOB_ID: 更新するジョブの ID と同じジョブ ID。
   • JOB_NAME: 更新するジョブの名前と同じジョブ名。

   パイプライン内の変換名が変更された場合は、変換マッピングを用意して transformNameMapping フィールドで渡す必要があります。

3. 省略可: curl（Linux、macOS、Cloud Shell）を使用してリクエストを送信するには、リクエストを JSON ファイルに保存し、次のコマンドを実行します。

   curl -X POST -d "@FILE_PATH" -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)"  https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs
   

   FILE_PATH は、リクエスト本文を含む JSON ファイルのパスに置き換えます。

Java

置換ジョブを起動するときに --jobName オプションで渡す値は、置き換えられるジョブの名前と厳密に一致する必要があります。

Python

置換ジョブを起動するときに --job_name オプションで渡す値は、置き換えられるジョブの名前と厳密に一致する必要があります。

Go

置換ジョブを起動するときに --job_name オプションで渡す値は、置き換えられるジョブの名前と厳密に一致する必要があります。

gcloud

置換ジョブを起動するとき、JOB_NAME は、置き換えられるジョブの名前と厳密に一致する必要があります。

REST

replaceJobId フィールドの値に、更新するジョブと同じジョブ ID を設定します。正しいジョブ名の値を見つけるには、Dataflow モニタリング インターフェースで前のジョブを選択します。次に、[ジョブ情報] サイドパネルで [ジョブ ID] フィールドを見つけます。

Java

マッピングは、次の一般的な形式で --transformNameMapping コマンドライン オプションを使用して渡します。

--transformNameMapping= . 
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transformNameMapping でマッピング エントリを提供する必要があるのは、前のパイプラインと置換パイプラインの間で変更された変換名についてのみです。

--transformNameMapping を指定して実行する場合、シェルに応じて引用符をエスケープする必要があります。次に Bash での例を示します。

--transformNameMapping='{"oldTransform1":"newTransform1",...}'

Python

マッピングは、次の一般的な形式で --transform_name_mapping コマンドライン オプションを使用して渡します。

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform_name_mapping でマッピング エントリを提供する必要があるのは、前のパイプラインと置換パイプラインの間で変更された変換名についてのみです。

--transform_name_mapping を指定して実行する場合、シェルに応じて引用符をエスケープする必要があります。次に Bash での例を示します。

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

Go

マッピングは、次の一般的な形式で --transform_name_mapping コマンドライン オプションを使用して渡します。

--transform_name_mapping= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform_name_mapping でマッピング エントリを提供する必要があるのは、前のパイプラインと置換パイプラインの間で変更された変換名についてのみです。

--transform_name_mapping を指定して実行する場合、シェルに応じて引用符をエスケープする必要があります。次に Bash での例を示します。

--transform_name_mapping='{"oldTransform1":"newTransform1",...}'

gcloud

マッピングは、次の一般的な形式で --transform-name-mappings オプションを使用して渡します。

--transform-name-mappings= .
{"oldTransform1":"newTransform1","oldTransform2":"newTransform2",...}

--transform-name-mappings でマッピング エントリを提供する必要があるのは、前のパイプラインと置換パイプラインの間で変更された変換名についてのみです。

--transform-name-mappings を指定して実行する場合、シェルに応じて引用符をエスケープする必要があります。次に Bash での例を示します。

--transform-name-mappings='{"oldTransform1":"newTransform1",...}'

REST

マッピングは、次の一般的な形式で transformNameMapping フィールドを使用して渡します。

"transformNameMapping": {
  oldTransform1: newTransform1,
  oldTransform2: newTransform2,
  ...
}

transformNameMapping でマッピング エントリを提供する必要があるのは、前のパイプラインと置換パイプラインの間で変更された変換名についてのみです。

Java

  .apply("FormatResults", ParDo
    .of(new DoFn<KV<String, Long>>, String>() {
      ...
     }
  }))

Python

  | 'FormatResults' >> beam.ParDo(MyDoFn())

Go

  // In Go, this is always the package-qualified name of the DoFn itself.
  // For example, if the FormatResults DoFn is in the main package, its name
  // is "main.FormatResults".
  beam.ParDo(s, FormatResults, results)

Java

--transformNameMapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

置換パイプラインで変換を完全に削除する場合は、null マッピングを提供する必要があります。置換パイプラインが CountWidgets/Parse 変換を完全に削除すると仮定します。

--transformNameMapping={"CountWidgets/Parse":""}

Python

--transform_name_mapping={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

置換パイプラインで変換を完全に削除する場合は、null マッピングを提供する必要があります。置換パイプラインが CountWidgets/Parse 変換を完全に削除すると仮定します。

--transform_name_mapping={"CountWidgets/Parse":""}

Go

--transform_name_mapping={"CountWidgets/main.Parse":"CountWidgets/Scan/main.Parse"}

置換パイプラインで変換を完全に削除する場合は、null マッピングを提供する必要があります。置換パイプラインが CountWidgets/Parse 変換を完全に削除すると仮定します。

--transform_name_mapping={"CountWidgets/main.Parse":""}

gcloud

--transform-name-mappings={"CountWidgets/Parse":"CountWidgets/Scan/Parse"}

置換パイプラインで変換を完全に削除する場合は、null マッピングを提供する必要があります。置換パイプラインが CountWidgets/Parse 変換を完全に削除すると仮定します。

--transform-name-mappings={"CountWidgets/main.Parse":""}

REST

"transformNameMapping": {
  CountWidgets/Parse: CountWidgets/Scan/Parse
}

置換パイプラインで変換を完全に削除する場合は、null マッピングを提供する必要があります。置換パイプラインが CountWidgets/Parse 変換を完全に削除すると仮定します。

"transformNameMapping": {
  CountWidgets/main.Parse: null
}

Java

  p.apply("Read", ReadStrings())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Format' >> FormatStrings()

Go

   beam.ParDo(s, ReadStrings)
   beam.ParDo(s, FormatStrings)

Java

  p.apply("Read", ReadStrings())
   .apply("Remove", RemoveStringsStartingWithA())
   .apply("Format", FormatStrings());

Python

  p | 'Read' >> beam.io.ReadFromPubSub(subscription=known_args.input_subscription)
    | 'Remove' >> RemoveStringsStartingWithA()
    | 'Format' >> FormatStrings()

Go

  beam.ParDo(s, ReadStrings)
  beam.ParDo(s, RemoveStringsStartingWithA)
  beam.ParDo(s, FormatStrings)

Java

制限により、コンソールまたはターミナルで更新の失敗に関するエラーを表示するには、ブロッキング実行を使用する必要があります。現在の回避策の手順は次のとおりです。

1. pipeline.run().waitUntilFinish() をパイプラインのコードの中で使用します。
2. 置換パイプライン プログラムを --update オプション付きで実行します。
3. 置換ジョブが互換性チェックに合格するまで待ちます。
4. Ctrl+C を押してブロッキング実行プロセスを終了します。

また、Dataflow モニタリング インターフェースで置換ジョブの状態をモニタリングすることもできます。ジョブが正常に起動した場合は、互換性チェックにも合格しています。

Python

制限により、コンソールまたはターミナルで更新の失敗に関するエラーを表示するには、ブロッキング実行を使用する必要があります。現在の回避策の手順は次のとおりです。

1. パイプラインのコードの中で pipeline.run().wait_until_finish() を使用します。
2. 置換パイプライン プログラムを --update オプション付きで実行します。
3. 置換ジョブが互換性チェックに合格するまで待ちます。
4. Ctrl+C を押してブロッキング実行プロセスを終了します。

また、Dataflow モニタリング インターフェースで置換ジョブの状態をモニタリングすることもできます。ジョブが正常に起動した場合は、互換性チェックにも合格しています。

Go

制限により、コンソールまたはターミナルで更新の失敗に関するエラーを表示するには、ブロッキング実行を使用する必要があります。具体的には、--execute_async フラグまたは --async フラグを使用して非ブロッキング実行を指定する必要があります。現在の回避策の手順は次のとおりです。

1. --update オプションを指定し、--execute_async フラグまたは --async フラグを指定せずに、置換パイプライン プログラムを実行します。
2. 置換ジョブが互換性チェックに合格するまで待ちます。
3. Ctrl+C を押してブロッキング実行プロセスを終了します。

gcloud

制限により、コンソールまたはターミナルで更新の失敗に関するエラーを表示するには、ブロッキング実行を使用する必要があります。現在の回避策の手順は次のとおりです。

1. Java パイプラインの場合は、pipeline.run().waitUntilFinish() をパイプラインのコードの中で使用します。Python パイプラインの場合は、パイプラインのコードの中で pipeline.run().wait_until_finish() を使用します。Go パイプラインの場合は、[Go] タブの手順に沿って操作します。
2. 置換パイプライン プログラムを --update オプション付きで実行します。
3. 置換ジョブが互換性チェックに合格するまで待ちます。
4. Ctrl+C を押してブロッキング実行プロセスを終了します。

REST

制限により、コンソールまたはターミナルで更新の失敗に関するエラーを表示するには、ブロッキング実行を使用する必要があります。現在の回避策の手順は次のとおりです。

• Java パイプラインの場合は、pipeline.run().waitUntilFinish() をパイプラインのコードの中で使用します。Python パイプラインの場合は、パイプラインのコードの中で pipeline.run().wait_until_finish() を使用します。Go パイプラインの場合は、[Go] タブの手順に沿って操作します。
• replaceJobId フィールドを指定して、置換パイプライン プログラムを実行します。
• 置換ジョブが互換性チェックに合格するまで待ちます。
• Ctrl+C を押してブロッキング実行プロセスを終了します。