ETL 測試策略建構強壯資料管線

– 13 分鐘閱讀

在資料工程領域，ETL 管線的穩定性至關重要。本文將探討如何透過不同層次的測試策略，包含單元測試、整合測試、端對端測試以及驗證測試，來確保 ETL 流程的品質與可靠性。此外，文章也將介紹效能測試和韌性測試，以提升管線的效率和容錯能力，並提供使用 pytest 框架編寫測試案例的 Python 程式碼範例，最後說明如何結合 CI/CD 工具，開發自動化的測試與佈署流程，提升開發效率。

ETL 測試策略：建立強壯的資料管線

在前面的章節中，我們探討了 ETL 管線的協調與擴充套件，認識到隨著資料量和複雜度的增加，高效且可擴充套件的管線變得越來越重要。在本章中，我們將重點轉移到 ETL 測試的不同策略和工具上，旨在進一步提升管線的效能和韌性。

技術需求

為了有效地利用本章提供的資源和程式碼範例，請確保您的系統符合以下技術需求：

  • 軟體需求：
    • 整合開發環境（IDE）：建議使用 PyCharm 作為主要的 IDE。
    • 安裝 Jupyter Notebooks。
    • 安裝 Python 3.6 或更高版本。
    • 安裝 Pipenv 以管理依賴項。
  • GitHub 儲存函式庫：本章相關的程式碼和資源可以在以下 GitHub 儲存函式庫中找到：https://github.com/PacktPublishing/Building-ETL-Pipelines-with-Python。

測試資料管執行緒式碼的好處

測試策略是成功佈署資料管線的幕後英雄。它們保障了流經管線的資料的品質、準確性和可靠性。徹底的測試可以提供對系統韌性的信心，並有助於高效地識別瓶頸和最佳化機會。

如何為您的 ETL 管線選擇合適的測試策略

下圖顯示了用於強壯資料管線的多個 ETL 測試領域：

@startuml
skinparam backgroundColor #F8F9FA
skinparam packageStyle rectangle
skinparam componentStyle rectangle

package "ETL 測試金字塔架構" {

  package "單元測試層 (70% 測試比例)" #E8F5E9 {
    component [Extract 函式\n單元測試] as ExtractTest
    component [Transform 邏輯\n單元測試] as TransformTest
    component [Load 函式\n單元測試] as LoadTest
    component [資料驗證\n函式測試] as ValidationTest
    component [錯誤處理\n測試] as ErrorTest
  }

  package "整合測試層 (20% 測試比例)" #FFF9C4 {
    component [資料庫連線\n整合測試] as DBIntegrationTest
    component [API 端點\n整合測試] as APIIntegrationTest
    component [ETL 管線\n元件整合] as PipelineIntegration
    component [外部服務\n整合測試] as ExternalServiceTest
  }

  package "端對端測試層 (8% 測試比例)" #FFE0B2 {
    component [完整 ETL\n流程測試] as E2ETest
    component [資料品質\n端對端驗證] as DataQualityE2E
    component [多資料源\n整合測試] as MultiSourceTest
  }

  package "效能測試層 (2% 測試比例)" #FFCDD2 {
    component [負載測試\n(Load Testing)] as LoadTestPerf
    component [壓力測試\n(Stress Testing)] as StressTest
    component [延遲測試\n(Latency Testing)] as LatencyTest
    component [吞吐量測試] as ThroughputTest
  }

  package "驗證測試 (貫穿各層)" #E1BEE7 {
    component [Schema 驗證] as SchemaValidation
    component [資料完整性\n檢查] as DataIntegrity
    component [商業邏輯\n驗證] as BusinessLogic
    component [資料一致性\n測試] as ConsistencyTest
  }
}

cloud "測試工具生態系" {
  [pytest] as pytest
  [pytest-cov] as coverage
  [Great Expectations] as ge
  [Apache Airflow\nTest Utils] as airflow
  database [測試資料庫\n(Test DB)] as testdb
  database [Mock 服務] as mock
}

' 單元測試層連接
ExtractTest --> pytest : 使用
TransformTest --> pytest : 使用
LoadTest --> pytest : 使用
ValidationTest --> ge : 資料驗證
ErrorTest --> pytest : 異常處理

' 整合測試層連接
DBIntegrationTest --> testdb : 連接測試DB
APIIntegrationTest --> mock : Mock API
PipelineIntegration --> airflow : Airflow測試
ExternalServiceTest --> mock : Mock外部服務

' 端對端測試層連接
E2ETest --> pytest : E2E框架
E2ETest --> airflow : DAG測試
DataQualityE2E --> ge : 資料品質檢查
MultiSourceTest --> testdb : 多源整合

' 效能測試層連接
LoadTestPerf --> pytest : 效能測試
StressTest --> coverage : 覆蓋率分析
LatencyTest --> coverage : 延遲監控
ThroughputTest --> pytest : 吞吐量測試

' 驗證測試連接
SchemaValidation --> ge : Schema驗證
DataIntegrity --> ge : 完整性檢查
BusinessLogic --> pytest : 邏輯驗證
ConsistencyTest --> ge : 一致性測試

' 層級之間的依賴關係
ExtractTest ..> DBIntegrationTest : 支援
TransformTest ..> PipelineIntegration : 支援
LoadTest ..> DBIntegrationTest : 支援

DBIntegrationTest ..> E2ETest : 支援
PipelineIntegration ..> E2ETest : 支援

E2ETest ..> LoadTestPerf : 基於
E2ETest ..> DataQualityE2E : 包含

note right of E2ETest
  **端對端測試流程:**
  1. 準備測試資料集
  2. 執行完整 ETL 管線
  3. 驗證輸出結果
  4. 檢查資料完整性
  5. 確認商業邏輯正確性
  6. 清理測試環境
end note

note left of ValidationTest
  **資料驗證檢查項目:**
  • Schema 結構驗證
  • 資料型別檢查
  • 範圍限制驗證
  • NULL 值處理規則
  • 重複資料檢測
  • 參照完整性檢查
  • 商業規則驗證
end note

note bottom of LoadTestPerf
  **效能測試關鍵指標:**
  • 處理延遲 (Latency)
  • 吞吐量 (Throughput)
  • 資源使用率 (CPU/Memory)
  • 並發處理能力
  • 錯誤率 (Error Rate)
end note

note top of ExtractTest
  **單元測試最佳實踐:**
  • 快速執行 (< 100ms)
  • 獨立性 (Isolated)
  • 可重複性 (Repeatable)
  • 高覆蓋率 (> 80%)
  • 使用 Mock 隔離依賴
end note

@enduml

內容解密：

此圖示完整呈現 ETL 測試策略的金字塔架構，將測試分為五大層次並展現各層次的測試比例與工具整合。單元測試層佔比最高（70%），針對 Extract、Transform、Load 各個獨立函式進行測試，確保每個元件的基礎功能正確，搭配資料驗證與錯誤處理測試，使用 pytest 框架執行，強調快速執行（< 100ms）、獨立性與高覆蓋率（> 80%）。整合測試層（20%）驗證不同元件之間的協作，包含資料庫連線測試、API 端點整合、ETL 管線元件整合及外部服務整合，透過測試資料庫與 Mock 服務隔離外部依賴，並整合 Apache Airflow 進行 DAG 測試。端對端測試層（8%）執行完整的 ETL 流程測試，從資料來源到目標系統進行全流程驗證，包含資料品質檢查與多資料源整合測試，測試流程涵蓋準備測試資料、執行 ETL、驗證輸出、檢查完整性、確認商業邏輯及清理環境六大步驟。效能測試層（2%）評估系統在不同負載下的表現，包含負載測試、壓力測試、延遲測試及吞吐量測試，關鍵指標包含處理延遲、吞吐量、資源使用率、並發處理能力與錯誤率。驗證測試作為橫切關注點貫穿各層，使用 Great Expectations 進行 Schema 驗證、資料完整性檢查、商業邏輯驗證與資料一致性測試，檢查項目涵蓋結構驗證、型別檢查、範圍限制、NULL 值處理、重複檢測、參照完整性及商業規則驗證。圖中虛線展示層級之間的支援依賴關係，單元測試支援整合測試，整合測試支援端對端測試，端對端測試則作為效能測試的基礎。整體架構充分體現了測試金字塔理論，底層測試數量多且執行快速，頂層測試數量少但覆蓋範圍廣，搭配完整的測試工具生態系（pytest、pytest-cov、Great Expectations、Apache Airflow），確保 ETL 管線的正確性、可靠性、效能與可維護性。

此圖示展示了多個 ETL 測試領域之間的關係。在選擇適合您的 ETL 管線的測試策略時，需要考慮以下因素：

  • 專案範圍：小型專案可能只需要單元測試和基本的整合測試，而大型專案則需要更全面的方法，包括端對端測試、驗證測試和效能測試。
  • 資料敏感性：如果您正在處理敏感或受監管的資料，則可能需要專門的測試來確保符合相關法律法規的要求。
  • 團隊技能：您的團隊的能力也可以指導您選擇測試策略。某些策略需要專門的技能，如效能測試、資料驗證或安全測試。
  • 預算和時間：全面的測試策略提供更好的覆寫範圍，但也需要更多時間和資源。
  • 資料性質：如果您的管線處理多種資料型別、結構和大小，則需要多樣化的測試來確保所有資料都被準確處理。

建立 ETL 管線的測試生態系統

在建立 ETL 管線的測試生態系統時，需要考慮以下幾個方面：

  • 使用 pytest 等測試框架來編寫單元測試和整合測試。
  • 使用 Jupyter Notebooks 來進行互動式測試和除錯。
  • 使用 Pipenv 來管理依賴項和確保環境的一致性。

ETL 管線測試的最佳實踐

以下是一些 ETL 管線測試的最佳實踐：

  • 編寫全面的單元測試和整合測試，以確保管線的每個元件都正常運作。
  • 使用端對端測試來驗證整個管線的功能。
  • 使用驗證測試來確保資料的準確性和一致性。
  • 使用效能測試來最佳化管線的效能。

ETL 管線測試的挑戰

在進行 ETL 管線測試時，可能會遇到以下挑戰：

  • 資料品質問題：資料品質問題可能會導致測試失敗或產生不正確的結果。
  • 環境差異：不同環境之間的差異可能會導致測試結果不一致。
  • 依賴項管理：管理依賴項可能會變得複雜，特別是在大型專案中。

程式碼範例：使用 pytest 編寫單元測試

import pytest
from etl_code import etl_task

def test_etl_task():
    # 準備測試資料
    input_data = [...]
    expected_output = [...]

    # 執行 ETL 任務
    output = etl_task(input_data)

    # 驗證輸出結果
    assert output == expected_output

內容解密：

  1. 匯入必要的模組：我們匯入了 pytest 來使用其測試功能，以及 etl_code 模組中的 etl_task 函式進行 ETL 處理。
  2. 定義單元測試函式：test_etl_task 函式是一個單元測試案例，用於驗證 etl_task 函式的功能是否正確。
  3. 準備測試資料：在這個範例中，我們假設 input_data 是輸入資料，而 expected_output 是預期的輸出結果。
  4. 執行 ETL 任務：我們呼叫 etl_task 函式並將 input_data 作為引數傳遞，得到輸出結果 output。
  5. 驗證輸出結果：使用 assert 陳述來檢查 output 是否等於 expected_output，以此來驗證 ETL 任務是否正確執行。

透過這種方式，我們可以確保 ETL 管線中的每個元件都經過充分的測試，從而提高整個管線的可靠性和穩定性。

測試 ETL 管道的策略

在開發和維護 ETL（提取、轉換、載入）管道時，測試是確保資料處理流程正確性和可靠性的關鍵步驟。有效的測試策略可以幫助您在早期發現問題，減少下游錯誤，並提高整體資料品質。

測試頻率

測試 ETL 管道的頻率取決於測試的型別和專案的需求。以下是一些常見的測試頻率：

  • 持續測試：對於活躍的專案，特別是在敏捷開發環境中，應盡可能頻繁地執行單元測試和小型整合測試，理想情況下，每次提交程式碼時都進行測試。
  • 排程測試：端對端測試和效能測試通常更耗時和昂貴，因此可能會安排在夜間或其他低活動期間執行。
  • 條件測試：某些測試可能會根據特定條件執行，例如在一定數量的新提交後或手動啟動時。
  • 資料架構變更後：如果源或目標資料函式庫的架構發生變化，強烈建議完整執行所有測試。
  • 定期監控：使用 Prometheus、Grafana 或 Datadog 等工具持續監控管道，並在檢測到異常或錯誤時觸發某些測試。

建立簡單的 ETL 管道測試

讓我們建立一個簡單的 ETL 管道，並為其建立測試。首先，開啟與本章相關的 Jupyter Notebook（Testing_Strategies_ETL_Pipelines.ipynb），檢視第一個單元格中的小型管道：

ETL 管道範例程式碼

def extract(file_path):
    with open(file_path, 'r') as file:
        data = file.read()
    return {'value': int(data)}

def transform(data):
    if data['value'] < 0:
        raise ValueError('Value must be positive.')
    data['value'] *= 2
    return data

def load(data, database):
    database['value'] = data['value']

單元測試

單元測試旨在驗證個別函式的功能是否符合預期。以下是使用 pytest 進行單元測試的範例：

def test_transform_unittest():
    input_data = {'value': 5}
    expected_output = {'value': 10}
    result = transform(input_data)
    assert result == expected_output, f'Expected {expected_output}, but got {result}'

內容解密：

此單元測試驗證 transform 函式是否正確地將輸入資料的 value 欄位乘以 2。首先定義輸入資料和預期輸出，然後呼叫 transform 函式，並斷言結果是否符合預期。

驗證測試

驗證測試旨在確保管道生成的資料符合預期結果。以下是使用 pytest.raises 進行驗證測試的範例：

def test_transform_validation():
    input_data = {'value': -5}
    with pytest.raises(ValueError) as excinfo:
        transform(input_data)
    assert str(excinfo.value) == 'Value must be positive.'

內容解密：

此驗證測試驗證當輸入資料的 value 欄位為負數時，transform 函式是否會引發 ValueError。使用 pytest.raises 捕捉異常，並斷言異常資訊是否符合預期。

整合測試

整合測試旨在驗證管道中不同元件之間的互動是否正確。以下是整合測試的範例：

def test_load_transform_integration():
    input_data = {'value': 5}
    expected_output = {'value': 10}
    database = {}
    transformed_data = transform(input_data)
    load(transformed_data, database)
    assert database == expected_output, f'Expected {expected_output}, but got {database}'

內容解密：

此整合測試驗證 transform 和 load 函式是否能夠正確地協同工作，將資料從源頭轉換並載入到目標資料函式庫中。首先定義輸入資料和預期輸出，然後呼叫 transform 和 load 函式，並斷言目標資料函式庫中的資料是否符合預期。

端對端測試

端對端測試旨在驗證整個 ETL 管道是否能夠正確地處理資料，從源頭到目標。以下是端對端測試的範例：

def test_pipeline_end_to_end():
    test_input_file = 'test_input.txt'
    expected_output = {'value': 20}
    with open(test_input_file, 'w') as file:
        file.write('10')
    input_data = extract(test_input_file)
    transformed_data = transform(input_data)
    database = {}
    load(transformed_data, database)
    assert database == expected_output, f'Expected {expected_output}, but got {database}'

內容解密：

此端對端測試驗證整個 ETL 管道是否能夠正確地處理資料，從源頭（檔案）到目標（資料函式庫）。首先建立一個測試輸入檔案，然後呼叫 extract、transform 和 load 函式，並斷言目標資料函式庫中的資料是否符合預期。

效能測試

效能測試旨在評估 ETL 管道的效能和可擴充套件性。以下是效能測試的範例：

import time

def test_transform_performance():
    input_data = [{'value': i} for i in range(1000000)]
    start_time = time.time()
    for data in input_data:
        transform(data)
    end_time = time.time()
    elapsed_time = end_time - start_time
    print(f"Elapsed time for processing 1 million data points was {elapsed_time} seconds.")

內容解密：

此效能測試評估 transform 函式處理大量資料（100 萬個資料點）的效能。首先定義輸入資料，然後記錄處理資料所需時間，並列印結果。

綜上所述，ETL 管道的測試策略應包括單元測試、驗證測試、整合測試、端對端測試和效能測試，以確保管道的正確性、可靠性和效能。透過實施這些測試，可以及早發現問題，提高資料品質，並確保 ETL 管道的平穩執行。

測試 ETL 管道的策略

176

韌性測試（Resilience Testing）

韌性測試的目的是確保系統能夠處理和從錯誤中還原。這涉及一系列測試，從檢查系統是否能優雅地處理錯誤輸入，到驗證系統在模擬故障後能否還原運作。

考慮一個韌性測試，模擬在轉換步驟中出現的暫時性錯誤，例如暫時性的網路中斷或存取遠端資源時的超時錯誤：

import random
def transform(input_data):
    """
    將輸入值加倍的轉換函式。
    假設 input_data 是一個具有 'value' 鍵的字典。
    """
    # 有 10% 的機率引發 TimeoutError
    if random.random() < 0.1: 
        raise TimeoutError('暫時性的網路中斷。')
    output_data = {'value': input_data['value'] * 2}
    return output_data

內容解密：

  • random.random() < 0.1 表示有 10% 的機率會引發 TimeoutError，模擬網路問題。
  • TimeoutError 是 Python 的內建異常類別，用於表示操作超時。
  • 此函式主要用於測試轉換步驟的可靠性。

我們可以編寫一個韌性測試，重試 transform() 函式直到成功或達到最大嘗試次數（例如五次）：

def test_transform_resilience_timeout_retry5():
    # 設定
    input_data = {'value': 5}
    expected_output = {'value': 10}
    
    # 執行
    for i in range(5):
        try:
            result = transform(input_data)
            break
        except TimeoutError:
            if i == 4: 
                raise 
    else:
        raise ValueError("轉換函式在 5 次嘗試後仍然失敗。")

    # 驗證
    assert result == expected_output, f'預期 {expected_output}，但得到 {result}'

內容解密：

  • 使用 for 迴圈重試 transform() 函式，最多嘗試五次。
  • 若 transform() 引發 TimeoutError，捕捉該異常並重試。
  • 若五次嘗試後仍然失敗，則引發 ValueError 表示測試失敗。
  • 使用 assert 陳述式驗證結果是否符合預期。

ETL 管道測試環境的最佳實踐

177

定義測試目標

在編寫任何程式碼之前，必須確定測試的目的和目標。為什麼需要在管道中進行測試？希望透過測試實作什麼目標？這些目標可能包括驗證資料完整性、確認資料轉換的準確性、驗證業務規則或檢查管道的效能和韌性。

建立測試框架

選擇一個符合技術堆積疊的測試框架。對於 Python，可以選擇 pytest 或 unittest。測試框架應該平衡單一測試與提供設定、拆卸和隔離測試的工具。

各類別測試的設計結構

單元測試（Unit Tests）

單元測試是測試資料流動的最小建構區塊。一個純粹的單元測試驗證管道中的一個可測試方面。用於捕捉開發過程中的問題，使除錯更容易。

積合測試（Integration Tests）

積合測試是複雜性的下一層。它們確保為單元測試建立的元件之間的資料流動正常運作。如果某個函式與另一個函式之間的互動出現問題，積合測試應該能夠偵測到。

端對端測試（End-to-End Tests）

端對端測試是與資料流動相關的最後一層複雜性。它們驗證整個管道是否按預期工作，有助於捕捉在隔離測試元件時不明顯的問題。

驗證測試（Validation Tests）

驗證測試確保管道的輸出結果符合預期。它們確保管道產生的資料的準確性和正確性，從而與下游使用者建立信任。

效能和韌性測試（Performance and Resilience Tests）

這類別測試幫助識別管道中的瓶頸或效能問題，並確保管道能夠從錯誤或故障中還原。

自動化 ETL 測試

一旦建立了測試，就應該自動化它們。利用 CI/CD 工具自動觸發建置、測試和佈署過程。例如，當程式碼變更被推播或合併到指定的分支時，自動執行測試。

圖 13.3 展示了 CI/CD 工具如何應用和佈署：

@startuml
:Repository; --> [Push Code] :Push;
:B; --> [Test] :Test;
:C; --> [Build & Package] :Build;
:D; --> [Deploy] :Deploy;
@enduml

此圖示說明瞭 CI/CD 工具如何在工作流程中發揮作用，從掃描和批准新程式碼到自動佈署。

常見的 CI/CD 工具

  • Jenkins：一個開源自動化伺服器，用於建置、測試和佈署軟體。
  • CircleCI：一個雲端的 CI/CD 平台，提供自動化測試和佈署功能。
  • GitHub Actions：GitHub 提供的 CI/CD 工具，允許在 GitHub 倉函式庫中自動化軟體建置、測試和佈署工作流程。

這些工具可以與 Docker 和 Kubernetes 等佈署自動化工具整合，以協調佈署過程。

玄貓 BlackCat

玄貓 BlackCat

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