更新現有管道

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 範本工作的作業圖，請使用 gcloud dataflow flex-template run 指令搭配 additional-experiments 選項：

• 傳送 --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 更新彈性範本作業，請使用 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 資源。加入 view 查詢參數，並將值設為 JOB_VIEW_DESCRIPTION。加入 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 相同。
   • 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 監控介面中選取先前的工作。然後在「Job info」側邊面板中，找到「Job 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 以結束封鎖執行器程序。