Databricks Lakehouse 資料管線監控實戰

– 11 分鐘閱讀

在資料工程領域，Databricks Lakehouse 已成為構建資料管線的重要平臺。本文將深入探討如何利用 Lakehouse 監控 Delta 表格的資料品質，並分享在生產環境中處理失敗情況的最佳實踐。同時，我們將介紹如何使用 Databricks Auto Loader 進行資料匯入，並示範如何設定執行時間門檻和 Webhook 警示，以實作自動化的監控和通知機制，確保資料管線的穩定性和資料品質。此外，文章也將探討 Delta Live Tables (DLT) 與 Unity Catalog 的整合應用，提供程式碼範例與最佳實踐，協助開發者構建更強健的資料處理流程。

Databricks Lakehouse 資料管線監控實戰

在資料驅動的時代，確保資料管線的穩定性和資料品質是至關重要的。Databricks Lakehouse 提供了一套完善的工具和框架，協助開發者構建可靠的資料處理流程。本文將深入探討如何在 Databricks 平臺上建立 Lakehouse Monitor，以及如何處理生產環境中的失敗情況。同時，我們將介紹 Databricks Auto Loader 在資料匯入方面的應用，並探討如何設定執行時間門檻和 Webhook 警示，以實作自動化的監控和通知。

提升生產環境中資料管線的監控能力

在生產環境中監控資料管線是確保資料品質和及時發現問題的關鍵。本章節將詳細介紹如何在 Databricks 平臺上建立 Lakehouse Monitor 來監控 Delta 表格，並探討如何在生產環境中處理失敗的情況。

建立 Lakehouse Monitor

Lakehouse Monitor 是 Databricks 提供的一個強大工具，用於監控 Delta 表格的資料品質。首先，我們需要在 Unity Catalog 中指定一個位置來儲存監控指標表格。例如，我們可以指定 chp10.monitor_demo 作為儲存指標的 catalog 和 schema 名稱。

設定 Lakehouse Monitor 的步驟：

  1. 在 Databricks 工作區中，導航至 Catalog Explorer。
  2. 選擇要監控的 Delta 表格，並點選 “Quality” 標籤頁。
  3. 點選 “Create monitor” 按鈕來建立 Lakehouse Monitor。
  4. 在 Metrics 部分，輸入指定的 catalog 和 schema 名稱（例如 chp10.monitor_demo）。
  5. 接受預設的工作區位置來儲存生成的儀錶板。

完成這些步驟後，Lakehouse Monitor 就會被建立。由於我們沒有設定排程，因此需要手動執行指標收集。回到 Catalog Explorer 中的 Delta 表格 “Quality” 標籤頁，點選 “Refresh metrics” 按鈕來手動觸發指標收集。

指標收集完成後：

  • 點選 “View dashboard” 按鈕來檢視捕捉的指標。
  • 這樣就成功建立了第一個 Lakehouse Monitor，為實作強健和自動化的資料品質可觀察性邁出了重要一步。

生產環境中的失敗處理最佳實踐

DLT（Delta Live Tables）框架在設計時就考慮到了失敗處理的問題。它能夠自動回應三種常見的管線失敗情況：

  • Databricks Runtime 迴歸
  • 更新處理失敗
  • 資料交易失敗

處理管線更新失敗

DLT 框架具備強大的錯誤處理機制。在管線更新過程中，如果發生處理錯誤，框架會將錯誤分類別為可重試錯誤或不可重試錯誤。預設情況下，DLT 框架會在檢測到可重試錯誤時重試管線更新兩次。

從表格交易失敗中還原

由於 Delta Lake 交易日誌的特性，對資料集的更改是原子的。這意味著如果交易在執行過程中失敗，整個交易會被放棄，從而防止資料集進入不確定的狀態。

實作練習：為超時任務設定 Webhook 警示

在本練習中，我們將建立一個自定義的 HTTP webhook，當 Databricks 中的排程任務執行時間超出預期時通知 HTTP 端點。

步驟：

  1. 在 Databricks 工作區中，建立一個新的 workflow，命名為 “Production Monitoring Demo”。
  2. 新增兩個任務：第一個任務使用 04a-IoT Device Data Generator.py notebook 生成輸入資料集；第二個任務執行 DLT 管線，使用 04b-IoT Device Data Pipeline.py notebook 讀取生成的資料。
  3. 為 workflow 設定超時閾值（例如120 分鐘），當執行時間超出此閾值時觸發通知。
  4. 建立一個新的 Webhook 目的地，輸入端點 URL、使用者名稱和密碼（如有需要）。
  5. 將 webhook 設定為當 workflow 超時或失敗時的通知目的地。

圖表翻譯：

此圖示展示瞭如何在 Databricks 工作區中建立一個 workflow 並設定相關的 DLT 管線更新。首先，我們建立一個新的 workflow，接著新增兩個任務：一個用於生成 IoT 裝置資料，另一個用於執行 DLT 管線更新。然後，我們為 workflow 設定超時閾值，並建立一個 Webhook 通知，以便在 workflow 超時或失敗時接收通知。

程式碼範例與解析

以下是一個簡單的 Python 程式碼範例，用於在 Databricks 中生成 IoT 裝置資料：

import pandas as pd
import numpy as np

# 生成模擬的 IoT 裝置資料
def generate_iot_data(num_records):
 data = {
 'device_id': np.arange(num_records),
 'temperature': np.random.uniform(20,30, num_records),
 'humidity': np.random.uniform(60,80, num_records)
 }
 return pd.DataFrame(data)

# 生成1000 筆記錄
iot_data = generate_iot_data(1000)
print(iot_data.head())

內容解密：

  1. 我們首先匯入必要的函式庫：pandas 用於資料處理，numpy 用於數值計算。
  2. 定義函式 generate_iot_data，它接受一個引數 num_records，代表要生成的記錄數量。
  3. 在函式內部，建立一個字典 data，包含三個鍵：device_id、temperature 和 humidity。
    • device_id 使用 np.arange(num_records) 生成從0 到 num_records-1 的連續整數。
    • temperature 和 humidity 分別使用 np.random.uniform 生成在指定範圍內的隨機浮點數。
  4. 使用字典 data 建立一個 pandas DataFrame，並傳回它

內容解密：

  1. infer data 函式生成模擬的 IoT 裝置資料，並傳回一個 DataFrame。
  2. 函式內部建立一個字典 data，包含裝置 ID、溫度和濕度三個欄位。
  3. 使用 np.arange 生成連續的裝置 ID，並使用 np.random.uniform 生成隨機的溫度和濕度資料。
  4. 最後，呼叫 generate_iot_data(1000) 生成1000 筆 IoT 裝置資料記錄，並列印出前幾筆記錄。

透過本章節的學習，我們瞭解瞭如何在 Databricks 平臺上建立 Lakehouse Monitor，以及如何處理生產環境中的失敗情況。同時，透過實作練習，我們掌握了為超時任務設定 Webhook 警示的方法，從而提升了資料管線的監控能力和可靠性。

監控資料管道的生產環境

設定執行時間門檻

  1. 導航至工作流程的 工作 分頁。
  2. 選擇特定的任務來設定門檻。
  3. 設定 最長執行時間 門檻為120 分鐘。
  4. 點選 儲存 按鈕以完成設定。

圖表說明：工作流程任務的執行時間門檻設定

圖表翻譯：

此圖表詳細展示了在 Databricks 平臺中如何針對特定工作流程任務設定執行時間門檻。圖中呈現了導航至工作分頁、選擇任務以及設定最長執行時間的步驟。

自動化監控與通知

當設定的執行時間門檻被觸發時，工作流程將自動停止，並傳送通知訊息至設定的 HTTP webhook 目的地，狀態訊息為 Timed Out。

資料品質監控
  1. 管道事件日誌：使管道擁有者能夠查詢資料管道的健康狀況、稽核性和效能，以及實時資料品質。
  2. Lakehouse 監控：自動監控資料集的統計指標，並在門檻被突破時通知團隊成員。
  3. 資料品質評估：透過管道評估資料品質，防止下游錯誤和不準確性。

重點技術與工具

  • Databricks Unity Catalog：實作資料治理，提供資料目錄、資料血緣和細粒度資料存取控制。
  • Delta Live Tables (DLT)：簡化資料管道的建立和維護，支援自動最佳化和管理。
  • Lakehouse 監控：提供強大的資料監控功能，確保資料品質和及時通知。

最佳實踐與未來方向

  • 持續整合和持續佈署 (CI/CD)：使用 Databricks Asset Bundles (DABs) 和 GitHub Actions 簡化佈署流程，實作自動化測試和佈署。
  • 資料血緣和稽核：利用 Unity Catalog 的資料血緣功能，追蹤資料來源和變更歷史，確保資料透明度和可稽核性。

程式碼範例：使用 Databricks Auto Loader 匯入資料

# 從雲端儲存中匯入 CSV 格式的資料
from pyspark.sql.functions import col

# 設定 Auto Loader 讀取 CSV 檔案
df = spark.readStream.format("cloudFiles") \
 .option("cloudFiles.format", "csv") \
 .option("header", "true") \
 .schema("device_id long, temperature double, humidity double") \
 .load("s3://example-bucket/data")

# 轉換資料
transformed_df = df.select(col("device_id"), col("temperature"), col("humidity"))

# 將資料寫入 Delta Lake
transformed_df.writeStream.format("delta") \
 .option("mergeSchema", "true") \
 .option("checkpointLocation", "s3://example-bucket/checkpoint") \
 .save("s3://example-bucket/delta_table")

內容解密：

此程式碼範例展示瞭如何使用 Databricks Auto Loader 從雲端儲存中匯入 CSV 格式的資料。首先，我們設定 Auto Loader 讀取 CSV 檔案，並指定 schema 的儲存位置。接著，我們對讀取的資料進行轉換，選擇需要的欄位。最後，將轉換後的資料寫入 Delta Lake 表格中，並啟用 schema 合併功能，以處理可能的 schema 變更。同時，我們指定了 checkpoint 的儲存位置，以確保資料處理的一致性和可靠性。

圖表翻譯：

此圖表呈現了使用 Databricks Auto Loader 和 Delta Lake 處理資料的流程。首先從雲端儲存讀取資料，接著進行資料轉換，最後將處理後的資料寫入 Delta Lake 表格，完成整個資料處理流程。

Delta Live Tables (DLT) 與 Unity Catalog 的整合應用

前言

Delta Live Tables (DLT) 是 Databricks 提供的一種用於構建和管理資料管道的框架，而 Unity Catalog 則是 Databricks 的統一資料治理解決方案。本文將探討 DLT 與 Unity Catalog 的整合應用，涵蓋其架構、功能以及最佳實踐。

DLT 基礎架構與概念

DLT 是根據 Delta Lake 構建的資料管道框架，支援流處理和批處理，具有以下關鍵概念：

  1. 資料處理模式：DLT 支援連續和觸發兩種處理模式，能夠靈活應對不同的業務需求。
  2. 資料品質監控：透過定義預期（expectations）來監控資料品質，並可根據預期結果採取相應的動作。
  3. 自動化表維護：DLT 自動執行表維護任務，如自動壓縮和清理過期資料。

程式碼範例：建立 DLT 管道

import dlt

@dlt.table(
 name="example_table",
 comment="An example table created using DLT"
)
def example_table():
 return spark.read.format("delta").load("path_to_delta_table")

內容解密：

  1. 使用 dlt.table 裝飾器定義一個 DLT 表。
  2. name 引數指定表的名稱，comment 引數提供表的描述資訊。
  3. 函式 example_table 傳回一個 DataFrame，該 DataFrame 從指定的 Delta Lake 表中讀取資料。

Unity Catalog 的架構與功能

Unity Catalog 是 Databricks 的統一資料治理解決方案，提供以下功能：

  1. 資料治理：統一管理和控制資料存取許可權。
  2. 資料血緣追蹤：追蹤資料的來源和去向，支援資料血緣分析。
  3. 資料儲存管理：支援多種儲存選項，包括內部儲存和外部儲存。

Unity Catalog 架構圖

圖表翻譯：

此圖示展示了 Unity Catalog 的基本架構。使用者透過 Unity Catalog 請求資料，Unity Catalog 驗證使用者許可權後，從內部或外部儲存中檢索資料並傳回給使用者。

DLT 與 Unity Catalog 的整合

將 DLT 與 Unity Catalog 整合，可以實作以下好處：

  1. 統一資料治理：透過 Unity Catalog 對 DLT 管道中的資料進行統一管理和控制。
  2. 資料血緣追蹤：利用 Unity Catalog 的資料血緣追蹤功能，監控 DLT 管道中的資料流動。
  3. 簡化資料管理：透過 Unity Catalog 簡化 DLT 管道中的資料儲存和管理。

程式碼範例：在 DLT 中使用 Unity Catalog

import dlt

@dlt.table(
 name="unity_catalog_table",
 comment="A table managed by Unity Catalog"
)
def unity_catalog_table():
 return spark.read.format("delta").load("path_to_delta_table").write.format("delta").saveAsTable("unity_catalog_table")

內容解密：

  1. 使用 dlt.table 裝飾器定義一個由 Unity Catalog 管理的 DLT 表。
  2. 將 DataFrame 寫入 Unity Catalog 管理的 Delta Lake 表中。

最佳實踐

  1. 使用 Terraform 管理 DLT 和 Unity Catalog 資源：利用 Terraform 的基礎設施即程式碼（IaC）能力，實作對 DLT 和 Unity Catalog 資源的自動化管理。
  2. 實施嚴格的資料存取控制：透過 Unity Catalog 對 DLT 管道中的資料實施嚴格的存取控制，確保資料安全。
  3. 監控和最佳化 DLT 管道效能：利用 Databricks 提供的監控工具，對 DLT 管道進行效能監控和最佳化。

程式碼範例：使用 Terraform 組態 DLT 管道

resource "databricks_dlt_pipeline" "example_pipeline" {
 name = "example_pipeline"
 storage = "s3://example-bucket/storage"
 configuration = {
 "key" = "value"
 }
}

內容解密：

  1. 使用 Terraform 的 databricks_dlt_pipeline 資源建立一個 DLT 管道。
  2. 組態管道的儲存位置和其他相關組態。

透過本文的介紹，我們深入瞭解了如何在 Databricks 平臺上建立 Lakehouse Monitor，以及如何處理生產環境中的失敗情況。同時，我們探討了 Databricks Auto Loader 在資料匯入方面的應用，以及如何設定執行時間門檻和 Webhook 警示，以實作自動化的監控和通知。此外，我們還介紹了 DLT 與 Unity Catalog 的整合應用，涵蓋其架構、功能以及最佳實踐。這些技術和工具將幫助開發者構建更加可靠和高效的資料處理流程。

現代資料應用程式建置：Databricks Lakehouse 平臺深度解析

Databricks Lakehouse 平臺是當前資料處理領域的重要解決方案，其結合了資料湖（Data Lake）與資料倉儲（Data Warehouse）的優勢，為企業提供高效、靈活的資料管理能力。本文將深入探討 Databricks Lakehouse 的核心功能、技術優勢及其在現代資料應用程式開發中的實際應用。

Databricks Lakehouse 技術架構解析

Databricks Lakehouse 的技術架構建立在 Apache Spark 的基礎上，充分利用了 Spark 在大資料處理方面的強大能力。以下是其核心架構組成：

圖表翻譯：

此圖示展示了 Databricks Lakehouse 的整體技術架構。首先，資料從各種來源被擷取到系統中。接著，資料經過 Apache Spark 進行處理。處理後的資料儲存在 Lakehouse 結構中。最後，這些資料被用於分析，並進一步應用於商業智慧或機器學習等高階應用場景。整個流程體現了資料從原始狀態到產生商業價值的完整路徑。

資料處理核心：Apache Spark 深度應用

Apache Spark 是 Databricks Lakehouse 的核心處理引擎。以下是使用 Spark 進行資料處理的示例程式碼：

from pyspark.sql import SparkSession

# 初始化 SparkSession
spark = SparkSession.builder.appName("LakehouseExample").getOrCreate()

# 讀取資料
df = spark.read.format("delta").load("/path/to/delta-table")

# 資料轉換
df_filtered = df.filter(df["age"] > 30)

# 結果寫回 Lakehouse
df_filtered.write.format("delta").mode("overwrite").save("/path/to/output")

內容解密：

此段程式碼展示瞭如何使用 Apache Spark 與 Databricks Lakehouse 進行資料處理。首先，建立了一個 SparkSession，這是與 Spark 互動的主要入口。接著，使用 Delta 格式讀取資料。透過簡單的 filter 操作對資料進行篩選。最後，將處理後的結果以 Delta 格式寫回儲存系統。整個過程體現了 Spark 在大資料處理中的高效與靈活性。

Lakehouse 儲存架構優勢

Databricks Lakehouse 的儲存架構具有多項技術優勢：

  1. 開放式架構：根據開放標準，支援多種資料格式
  2. 高效能查詢：透過最佳化技術實作快速查詢回應
  3. ACID 交易支援：確保資料操作的原子性與一致性
  4. 無縫擴充套件：支援大規模資料儲存與處理需求

效能最佳化實踐

在實際應用中，Databricks Lakehouse 的效能最佳化至關重要。以下是一些常見的最佳化策略：

# 資料分割槽最佳化示例
df.write \
 .format("delta") \
 .partitionBy("date") \
 .mode("overwrite") \
 .save("/path/to/partitioned-table")

內容解密：

此程式碼展示瞭如何透過資料分割槽來最佳化儲存和查詢效能。透過將資料按日期（date）進行分割槽，可以顯著提高查詢特定時間範圍資料的效率。這種分割槽策略在處理大規模時間序列資料時特別有效，能夠減少查詢所需的資料掃描量，從而提升整體系統效能。

安全性考量

在建置現代資料應用程式時，安全性是不可忽視的重要環節。Databricks Lakehouse 提供了多層級的安全保護機制：

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title Databricks Lakehouse 資料管線監控實戰

package "系統架構" {
    package "前端層" {
        component [使用者介面] as ui
        component [API 客戶端] as client
    }

    package "後端層" {
        component [API 服務] as api
        component [業務邏輯] as logic
        component [資料存取] as dao
    }

    package "資料層" {
        database [主資料庫] as db
        database [快取] as cache
    }
}

ui --> client : 使用者操作
client --> api : HTTP 請求
api --> logic : 處理邏輯
logic --> dao : 資料操作
dao --> db : 持久化
dao --> cache : 快取

note right of api
  RESTful API
  或 GraphQL
end note

@enduml

圖表翻譯：

此圖示展示了 Databricks Lakehouse 的安全架構。首先，使用者請求需要透過身分驗證。驗證透過後，系統會進行存取控制檢查。資料在傳輸和儲存過程中均進行加密處理。最後，所有操作都會被記錄在稽核日誌中。這個多層級的安全架構確保了資料的安全性和合規性。

未來發展趨勢

隨著資料量的持續增長和業務需求的變化，Databricks Lakehouse 不斷演進以滿足新的挑戰。未來的發展趨勢包括：

  1. 更強大的 AI/ML 整合能力
  2. 增強的即時資料處理能力
  3. 更完善的資料治理功能

參考實作案例

在實際應用中，Databricks Lakehouse 已被多家企業成功佈署，用於解決各種複雜的資料處理場景。例如，在即時資料分析領域，Lakehouse 能夠支援高並發的查詢請求，同時保持資料的即時更新。在機器學習應用中，Lakehouse 提供了高效的資料準備和模型訓練環境。

最佳實踐建議

  1. 合理規劃資料分割槽策略，以提升查詢效能
  2. 充分利用 Delta Lake 的 ACID 特性，確保資料操作的可靠性
  3. 定期進行效能監控和最佳化，以保持系統最佳執行狀態

透過遵循這些最佳實踐，企業能夠更好地發揮 Databricks Lakehouse 的技術優勢，構建高效穩定的現代資料應用程式。

玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。