Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

停止執行中的 Dataflow 管道

如要停止 Dataflow 工作，請使用Google Cloud 控制台、Cloud Shell、安裝 Google Cloud CLI 的本機終端機，或 Dataflow REST API。

您可以透過下列三種方式停止 Dataflow 工作：

取消工作。這個方法適用於串流管道和批次管道。取消工作後，Dataflow 服務會停止處理所有資料，包括緩衝資料。詳情請參閱「取消工作」。
排空工作。這個方法僅適用於串流管道。排空工作可讓 Dataflow 服務完成緩衝資料的處理作業，同時停止擷取新資料。詳情請參閱「排空工作」。
強制取消工作。這個方法適用於串流管道和批次管道。強制取消作業後，Dataflow 服務會立即停止處理任何資料，包括緩衝資料。強制取消前，請先嘗試正常取消。強制取消僅適用於卡在一般取消程序中的工作。詳情請參閱「強制取消工作」。

取消工作後，就無法重新啟動。如果未使用 Flex 範本，可以複製已取消的管道，並從複製的管道啟動新工作。

停止串流管道前，請考慮建立工作快照。Dataflow 快照會儲存串流管道的狀態，因此您可以啟動新版 Dataflow 工作，不會遺失狀態。詳情請參閱「使用 Dataflow 快照」。

如果管道較為複雜，建議建立範本，並從範本執行工作。

您無法刪除 Dataflow 工作，但可以封存已完成的工作。所有已完成的工作 (包括封存工作清單中的工作)，都會在 30 天保留期限過後刪除。

取消 Dataflow 工作

取消工作後，Dataflow 服務會立即停止工作。

取消工作後會發生下列情況：

Dataflow 服務會停止所有資料擷取和資料處理作業。
Dataflow 服務會開始清除附加至工作的 Google Cloud資源。

這些資源可能包括關閉 Compute Engine 工作站執行個體，以及關閉與 I/O 來源或接收器的有效連線。

取消工作的重要資訊

取消工作會立即停止處理管道。
取消工作時，您可能會遺失傳輸中的資料。傳輸中資料是指已讀取但管道仍在處理的資料。
取消工作前，從管道寫入輸出接收器的資料可能仍可存取。
如果資料遺失不是問題，取消工作可確保與工作相關聯的Google Cloud 資源盡快關閉。

排空 Dataflow 工作

排空工作時，Dataflow 服務會以目前狀態完成工作。如要在關閉串流管道時避免資料遺失，最佳做法是排空工作。

排空工作時，會發生下列動作：

收到排空要求後，工作很快就會停止從輸入來源擷取新資料 (通常在幾分鐘內)。
Dataflow 服務會保留所有現有資源 (例如工作站執行個體)，以完成管道中所有緩衝資料的處理和寫入作業。
所有待處理的處理和寫入作業完成後，Dataflow 服務會關閉與工作相關聯的 Google Cloud 資源。

如要排空工作，Dataflow 會停止讀取新輸入內容、以無限大的事件時間戳記標記來源，然後透過管道傳播無限大時間戳記。因此，正在排空的管道可能會出現無限浮水印。

排空作業的重要資訊

批次管道不支援排空工作。
在所有處理和寫入作業完成前，管道仍會產生維護相關Google Cloud 資源的費用。
您可以更新正在排空的管道。如果管道停滯，請使用修正問題的程式碼更新管道，即可順利排空管道，且不會遺失資料。
你可以取消耗電的工作。
排空作業可能需要相當長的時間才能完成，例如管道有大量緩衝資料時。
如果串流管道包含 Splittable DoFn，您必須先截斷結果，才能執行排空選項。如要進一步瞭解如何截斷 Splittable DoFn，請參閱 Apache Beam 說明文件。
在某些情況下，Dataflow 工作可能無法完成排空作業。您可以查看工作記錄，找出根本原因並採取適當行動。

資料保留

Dataflow 串流可容許 worker 重新啟動，且發生錯誤時不會導致串流工作失敗。而是會重試，直到您採取動作 (例如取消或重新啟動工作) 為止。排空工作時，Dataflow 會繼續重試，這可能會導致管道停滯。在這種情況下，如要順利排空管道並避免資料遺失，請更新管道程式碼，修正造成問題的錯誤。
Dataflow 服務會永久提交訊息，之後才會確認訊息。舉例來說，使用 Kafka 時，您可以將這個程序視為安全地將訊息擁有權從 Kafka 移交給 Dataflow，進而消除資料遺失的風險。

停滯的工作

排空作業無法修正管道停滯問題。如果資料移動遭到封鎖，管道會在排空指令後停滯。如要解決管道停滯的問題，請使用 update 指令更新管道，並提供可解決問題的程式碼。您也可以取消停滯的工作，但取消工作可能會導致資料遺失。

計時器

如果串流管道程式碼包含迴圈計時器，工作可能會變慢或無法排空。由於所有計時器完成後，排空作業才會結束，因此具有無限迴圈計時器的管道永遠不會完成排空作業。
Dataflow 會等待所有處理時間計時器完成，而不是立即觸發，這可能會導致排空速度緩慢。

排除工作造成的影響

排空串流管道時，Dataflow 會立即關閉所有處理中的視窗，並觸發所有觸發條件。

系統不會在排空作業中，等待任何未完成的未結算時間範圍。

舉例來說，如果管道在兩小時的時段內執行了十分鐘，而您排空工作，Dataflow 就不會等待時段的剩餘時間結束。並立即關閉視窗，只顯示部分結果。 Dataflow 會將資料時間戳記推進至無限大，導致開啟的視窗關閉。這項功能也適用於自訂資料來源。

如果管道使用自訂資料來源類別，Dataflow 會停止發出新資料的要求、將資料時間戳記推進至無限大，並在最後一個檢查點呼叫來源的 finalize() 方法。

如果排水不良，窗戶可能會部分積水。在這種情況下，如果重新啟動已排空的管道，同一個視窗可能會再次觸發，進而導致資料發生問題。舉例來說，在下列情境中，檔案可能會有名稱衝突，且資料可能會遭到覆寫：

如果您在下午 12:34 耗盡管道，則下午 12:00 至下午 1:00 的時間範圍會關閉，且只會納入時間範圍前 34 分鐘內觸發的資料。管道不會讀取下午 12:34 後的新資料。

如果隨即重新啟動管道，系統會再次觸發下午 12:00 至下午 1:00 的時間範圍，但只會讀取下午 12:35 至下午 1:00 的資料。系統不會傳送重複檔案，但如果檔案名稱重複，資料就會遭到覆寫。

您可以在 Google Cloud 控制台中查看管道轉換的詳細資料。下圖顯示程序內排空作業的影響。請注意，浮水印會前移至最大值。

排空作業的步驟檢視畫面。

圖 1. 排空作業的步驟檢視畫面。

強制取消 Dataflow 工作

只有在無法使用其他方法取消作業時，才使用強制取消。強制取消會終止工作，但不會清除所有資源。如果重複使用強制取消功能，可能會累積洩漏的資源，而洩漏的資源會占用配額。

強制取消工作時，Dataflow 服務會立即停止工作，導致 Dataflow 工作建立的 VM 洩漏。強制取消前，請先嘗試正常取消，且必須至少間隔 30 分鐘。

強制取消工作時，會發生下列情況：

Dataflow 服務會停止所有資料擷取和資料處理作業。

強制取消工作的重要資訊

強制取消工作會立即停止處理管道。
強制取消工作僅適用於卡在一般取消程序中的工作。
Dataflow 工作建立的任何 worker 執行個體不一定會釋出，這可能會導致 worker 執行個體洩漏。洩漏的 worker 執行個體不會計入作業費用，但可能會使用配額。作業取消後，您就可以刪除這些資源。

如果是 Dataflow Prime 工作，您無法查看或刪除洩漏的 VM。在大多數情況下，這些 VM 不會造成問題。不過，如果洩漏的 VM 造成問題 (例如耗用 VM 配額)，請與支援團隊聯絡。

停止 Dataflow 工作

停止工作前，請務必瞭解取消、排空或強制取消工作會造成哪些影響。

控制台

前往 Dataflow 的「Jobs」(工作) 頁面：

前往工作
按一下要停止的工作。

如要停止工作，工作狀態必須為「執行中」。
在工作詳細資料頁面中，按一下「停止」。
執行下列其中一個步驟：
- 如果是批次管道，請按一下「取消」或「強制取消」。
- 如果是串流管道，請按一下「取消」、「排空」或「強制取消」。
按一下「停止工作」，確認要停止工作。

gcloud

如要排空或取消 Dataflow 工作，可以使用 Cloud Shell 或已安裝 gcloud CLI 的本機終端機中的 gcloud dataflow jobs 指令。

登入 Shell。
列出正在執行的 Dataflow 工作 ID，然後記下要停止的工作 ID：
```
gcloud dataflow jobs list
```
如未設定 --region 旗標，系統會顯示所有可用區域的 Dataflow 工作。
執行下列其中一個步驟：
- 如要排空串流工作：
```
gcloud dataflow jobs drain JOB_ID
```
  將 JOB_ID 替換成您先前複製的工作 ID。
- 如要取消批次或串流工作，請按照下列步驟操作：
```
gcloud dataflow jobs cancel JOB_ID
```
  將 JOB_ID 替換成您先前複製的工作 ID。
- 如要強制取消批次或串流工作，請按照下列步驟操作：
```
gcloud dataflow jobs cancel JOB_ID --force
```
  將 JOB_ID 替換成您先前複製的工作 ID。

API

如要使用 Dataflow REST API 取消或排空工作，可以選擇 projects.locations.jobs.update 或 projects.jobs.update。在要求主體中，於所選 API 的工作執行個體 requestedState 欄位中，傳遞必要工作狀態。

重要事項：建議使用 projects.locations.jobs.update，因為 projects.jobs.update 僅允許更新在 us-central1 中執行的工作狀態。

如要取消工作，請將工作狀態設為 JOB_STATE_CANCELLED。
如要排空工作，請將工作狀態設為 JOB_STATE_DRAINED。
如要強制取消工作，請將工作狀態設為 JOB_STATE_CANCELLED，並加上 "force_cancel_job": "true" 標籤。要求主體如下：
```
{
  "requestedState": "JOB_STATE_CANCELLED",
  "labels": {
    "force_cancel_job": "true"
  }
}
```

偵測 Dataflow 工作完成

如要偵測工作取消或排空作業何時完成，請使用下列其中一種方法：

使用 Managed Service for Apache Airflow 等工作流程自動化調度管理服務，監控 Dataflow 工作。
同步執行管道，以便封鎖工作，直到管道完成為止。詳情請參閱「設定管道選項」中的「控管執行模式」。
使用 Google Cloud CLI 中的指令列工具輪詢工作狀態。如要取得專案中的所有 Dataflow 工作清單，請在殼層或終端機中執行下列指令：
```
gcloud dataflow jobs list
```
輸出內容會顯示每個工作的 ID、名稱、狀態 (STATE) 和其他資訊。詳情請參閱「使用 Google Cloud CLI 列出作業」。

封存已完成的 Dataflow 工作

封存 Dataflow 工作後，該工作就會從控制台的 Dataflow「Jobs」(工作) 頁面工作清單中移除。工作會移至封存的工作清單。您只能封存已完成的工作，包括下列狀態的工作：

JOB_STATE_CANCELLED
JOB_STATE_DRAINED
JOB_STATE_DONE
JOB_STATE_FAILED
JOB_STATE_UPDATED

如要進一步瞭解如何驗證這些狀態，請參閱「偵測 Dataflow 工作完成狀態」。

如要瞭解如何排解封存工作時發生的問題，請參閱「排解 Dataflow 錯誤」一文中的「封存工作錯誤」。

封存的工作會在 30 天保留期限過後刪除。

封存工作

請按照下列步驟，從 Dataflow「工作」頁面的主要工作清單中移除已完成的工作。

控制台

前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

前往工作

系統會列出 Dataflow 工作及其狀態。
選取工作。
在「Job Details」(工作詳細資料) 頁面中，按一下「Archive」(封存)。如果工作尚未完成，就無法使用「封存」選項。

REST

如要使用 API 封存工作，請使用 projects.locations.jobs.update 方法。

在這項要求中，您必須指定更新後的 JobMetadata 物件。在 JobMetadata.userDisplayProperties 物件中，使用鍵/值組合 "archived":"true"。

除了更新 JobMetadata 物件，API 要求也必須在要求網址中加入 updateMask 查詢參數：

https://dataflow.googleapis.com/v1b3/[...]/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived

使用任何要求資料之前，請先修改下列項目的值：

PROJECT_ID：專案 ID
REGION：Dataflow 區域
JOB_ID：Dataflow 工作的 ID

HTTP 方法和網址：

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived

JSON 要求內文：

{
  "job_metadata": {
    "userDisplayProperties": {
      "archived": "true"
    }
  }
}

如要傳送要求，請選擇以下其中一個選項：

curl

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login 透過使用者帳戶登入 gcloud CLI，或是使用 Cloud Shell 自動登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X PUT \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived"

PowerShell

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PUT `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived" | Select-Object -Expand Content

您應該會收到如下的 JSON 回覆：

{
  "id": "JOB_ID",
  "projectId": "PROJECT_ID",
  "currentState": "JOB_STATE_DONE",
  "currentStateTime": "2025-05-20T20:54:41.651442Z",
  "createTime": "2025-05-20T20:51:06.031248Z",
  "jobMetadata": {
    "userDisplayProperties": {
      "archived": "true"
    }
  },
  "startTime": "2025-05-20T20:51:06.031248Z"
}

gcloud

這個指令會封存單一工作。工作必須處於最終狀態，才能封存。否則要求會遭拒。傳送指令後，請確認是否要運作執行。

使用下方的任何指令資料之前，請先替換以下項目：

JOB_ID：要封存的 Dataflow 工作 ID。
REGION：選用。工作的 Dataflow 區域。

執行下列指令：

Linux、macOS 或 Cloud Shell

gcloud dataflow jobs archive JOB_ID --region=REGION_ID

Windows (PowerShell)

gcloud dataflow jobs archive JOB_ID --region=REGION_ID

Windows (cmd.exe)

gcloud dataflow jobs archive JOB_ID --region=REGION_ID

您應該會收到類似以下的回應：

Archived job [JOB_ID].
createTime: '2025-06-29T11:00:02.432552Z'
currentState: JOB_STATE_DONE
currentStateTime: '2025-06-29T11:04:25.125921Z'
id: JOB_ID
jobMetadata:
userDisplayProperties:
    archived: 'true'
projectId: PROJECT_ID
startTime: '2025-06-29T11:00:02.432552Z'

查看及還原封存的工作

如要查看封存的工作，或將封存的工作還原至 Dataflow「Jobs」(工作) 頁面的主要工作清單，請按照下列步驟操作。

控制台

前往 Google Cloud 控制台的 Dataflow「Jobs」(工作) 頁面。

前往工作
按一下「已封存」切換按鈕。畫面上會顯示已封存的 Dataflow 工作清單。
選取工作。
如要將工作還原至 Dataflow「Jobs」(工作) 頁面的主要工作清單，請在「Job Details」(工作詳細資料) 頁面中按一下「Restore」(還原)。

REST

如要使用 API 還原已封存的工作，請使用 projects.locations.jobs.update 方法。

在這項要求中，您必須指定更新後的 JobMetadata 物件。在 JobMetadata.userDisplayProperties 物件中，使用鍵/值組合 "archived":"false"。

除了更新 JobMetadata 物件，API 要求也必須在要求網址中加入 updateMask 查詢參數：

https://dataflow.googleapis.com/v1b3/[...]/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived

使用任何要求資料之前，請先修改下列項目的值：

PROJECT_ID：專案 ID
REGION：Dataflow 區域
JOB_ID：Dataflow 工作的 ID

HTTP 方法和網址：

PUT https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived

JSON 要求內文：

{
  "job_metadata": {
    "userDisplayProperties": {
      "archived": "false"
    }
  }
}

如要傳送要求，請選擇以下其中一個選項：

curl

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

curl -X PUT \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived"

PowerShell

注意：下列指令假設您已執行 gcloud init 或 gcloud auth login，透過使用者帳戶登入 gcloud CLI。您可以執行 gcloud auth list，查看目前使用的帳戶。

將要求主體儲存在名為 request.json 的檔案中，然後執行下列指令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method PUT `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/jobs/JOB_ID/?updateMask=job_metadata.user_display_properties.archived" | Select-Object -Expand Content

您應該會收到如下的 JSON 回覆：

{
  "id": "JOB_ID",
  "projectId": "PROJECT_ID",
  "currentState": "JOB_STATE_DONE",
  "currentStateTime": "2025-05-20T20:54:41.651442Z",
  "createTime": "2025-05-20T20:51:06.031248Z",
  "jobMetadata": {
    "userDisplayProperties": {
      "archived": "false"
    }
  },
  "startTime": "2025-05-20T20:51:06.031248Z"
}

後續步驟

瀏覽 Dataflow REST API。
在 Google Cloud 控制台中探索 Dataflow 監控介面。
進一步瞭解如何更新管道。

停止執行中的 Dataflow 管道 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

取消 Dataflow 工作

取消工作的重要資訊

排空 Dataflow 工作

排空作業的重要資訊

資料保留

停滯的工作

計時器

排除工作造成的影響

強制取消 Dataflow 工作

強制取消工作的重要資訊

停止 Dataflow 工作

控制台

gcloud

API

偵測 Dataflow 工作完成

封存已完成的 Dataflow 工作

封存工作

控制台

REST

curl

PowerShell

gcloud

Linux、macOS 或 Cloud Shell

Windows (PowerShell)

Windows (cmd.exe)

查看及還原封存的工作

控制台

REST

curl

PowerShell

後續步驟

停止執行中的 Dataflow 管道