管線與作業中的設計錯誤處理

8 分鐘

小提示

有關更多詳細信息，請參閱 文本和圖像 選項卡！

當資料管線失效時，後果不僅僅停留在即時工作錯誤。無效紀錄會破壞下游分析，未被察覺的故障可能累積技術債務，若沒有適當的警示，團隊可能在問題發生數小時甚至數天後才發現。設計穩健的 錯誤處理 系統，能將您的流程從脆弱流程轉變為具備韌性的系統，保護資料品質並促進快速復原。

在這個單元中，你將學習如何設計 Azure Databricks 中資料管線、筆記本和工作中的錯誤處理策略。

了解錯誤處理的職責

Azure Databricks 中的錯誤處理在多個層級運作，每個層級針對不同的故障情境進行處理。平台會自動處理 基礎設施層級的問題 ，例如叢集節點故障。作為資料工程師，你的責任主要聚焦於 資料層級錯誤 與 應用邏輯故障。

以典型的 ETL 管線處理客戶交易為例。錯誤可能發生於以下情況：

來源資料包含格式錯誤或缺少必填欄位
轉換邏輯遇到意外值
外部服務連線逾時
運算資源變得無法使用。

每種情況都需要不同的回應。部分錯誤需立即終止管線，另一些則應記錄問題並繼續處理有效紀錄。

定義宣告式管線中的資料品質期望

Lakeflow 管線提供稱為期望的內建資料品質條件約束。這些限制會在資料流經管線時驗證紀錄，讓你掌控如何處理無效資料。

當紀錄驗證失敗時，有三個動作決定發生什麼事：

動作	行為	用例
警告	寫入目標的記錄無效；指標已記錄	監測資料品質趨勢
卸除	無效紀錄被排除在目標之外	防止不良資料傳播
Fail	管線立即停止	這些關鍵約束絕對不能被違反

小提示

當失敗預期觸發時，只有該特定流程會停止。同一管線中其他平行流程繼續處理。這種隔離性避免單一資料品質問題導致整個資料平台停擺。

設定工作層級錯誤處理

Lakeflow Jobs 提供多種機制來處理任務失敗並確保管線可靠性。這些設定協同運作，創造出具韌性的工作流程。

設定重試政策

像網路逾時、臨時資源限制或短暫服務中斷這類暫時性故障，通常會自行解決。設定 重試策略 ，自動再次嘗試失敗的任務：

重試次數：在標記任務失敗前的嘗試次數（通常為1至3次）
重試間隔：嘗試間隔，允許系統恢復

對於連續作業，平台會自動套用 指數退縮。每次連續失敗都會增加下一次重試前的等待時間，避免快速重複失敗導致資源過度消耗。

設定逾時

一個任務如果無限期卡住，會消耗資源並阻礙下游處理。設定 逾時閾值 以終止無反應任務：

預期持續時間：超過時會觸發警告通知
最大持續時間：超過任務時終止

對於串流任務，你也可以在任務設定面板的指標閾值下設定串流待辦清單的閾值。當串流來源的積壓超過你設定的閾值時，會觸發一個通知事件，使你能在延遲累積成更大的延遲之前及早察覺處理延遲。此功能目前處於公開預覽階段。

新增通知

主動警示能快速回應故障。透過以下方式設定通知以聯絡您的團隊：

電子郵件地址
Slack 管道
Microsoft 團隊
PagerDuty
自訂 webhook

你可以在工作開始、成功、失敗，或超過期限時觸發通知。

這很重要

每次重試都會觸發任務層級通知。為避免通知疲勞，設定任務通知時，請選擇 靜音通知直到最後一次重試 。

設計條件式任務流程

除了簡單的重試外，你還可以設計出能用 條件任務相依機制智能回應失敗的作業：

全部成功：只有當所有上游任務都成功完成後，才繼續
至少有一項失敗：觸發錯誤處理或清理任務
全部完成：不論上游狀態如何執行（對清理作業很有用）

此方法可實現如在處理失敗時執行通知任務，或在重試前執行清理邏輯等模式。

在筆記本中實作錯誤處理

用筆記本建管線時，你會直接在程式碼中實作錯誤處理。 Python 的 例外處理 提供對失敗回應的細緻控制。

捕捉並處理例外

將可能失敗的操作包裝在 try-except 區塊中：

from pyspark.errors import PySparkException

try:
    df = spark.read.table("source_data")
    transformed = df.filter("amount > 0")
    transformed.write.mode("append").saveAsTable("target_data")
except PySparkException as e:
    if e.getErrorClass() == "TABLE_OR_VIEW_NOT_FOUND":
        # Handle missing source table
        print(f"Source table not found: {e.getMessageParameters()['relationName']}")
    else:
        raise  # Re-raise unexpected errors

此範例捕捉了 PySparkException，它是 PySpark 錯誤的基底類別。該 getErrorClass() 方法會回傳標準化錯誤碼，例如 TABLE_OR_VIEW_NOT_FOUND，讓你能以不同方式處理特定錯誤。此 getMessageParameters() 方法提供錯誤的上下文，例如遺失資料表的名稱。你沒有明確處理的錯誤會被重新提出，以避免默默忽略意外問題。

實作重試邏輯

對於容易發生暫時性失敗的操作，請實作帶有指數退縮的重試：

import time

def run_with_retry(operation, max_retries=3, base_delay=10):
    for attempt in range(max_retries + 1):
        try:
            return operation()
        except Exception as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Attempt {attempt + 1} failed, retrying in {delay}s: {e}")
            time.sleep(delay)

向工作傳出成功或失敗的訊號

當筆記本作為工作任務運行時，請使用 dbutils.notebook.exit() 傳達結果。

try:
    # Processing logic
    records_processed = process_data()
    dbutils.notebook.exit(f"SUCCESS: Processed {records_processed} records")
except Exception as e:
    dbutils.notebook.exit(f"FAILED: {str(e)}")

下游任務則可在條件邏輯或報告中使用這些值。

應用錯誤處理最佳實務

有效的錯誤處理需要在全面性與實用性之間取得平衡。請參考以下指引：

重大錯誤的快速檢錯：當資料完整性受到威脅時，立即停止處理。持續使用損壞的資料會帶來更大的後續問題。

記錄有意義的背景：包含足夠資訊以診斷問題，同時避免暴露敏感資料。記錄下時間戳記、記錄次數與錯誤分類。

復原計畫：設計管線，使失敗的運行能被修復。盡可能使用冪等運算，並結構化作業，使只有失敗的任務需要重執行。

監控趨勢：追蹤資料品質指標。資料遺失的逐漸增加可能表示上游系統有變動需要關注。

測試失敗情境：透過在開發過程中故意引入失敗，驗證你的錯誤處理是否如預期般運作。

意見反應

此頁面對您有幫助嗎？