實施測試策略

8 分鐘

測試是可靠資料工程解決方案的關鍵基礎。當你實施 全面的測試策略時，能及早發現問題、驗證假設，並確保你的流程提供一致且值得信賴的結果。隨著資料量增加及管線複雜化， 自動化測試 成為維持品質與降低生產失敗風險的關鍵。

在本單元中，您將學習如何實作涵蓋 單元測試、 整合測試、 端對端測試及 Azure Databricks 使用者 驗收測試（UAT） 的測試策略。

了解考試金字塔

設計良好的測試策略遵循 測試金字塔 的概念。在基礎上，你會有許多快速且孤立的單元測試。往上看，驗證元件互動的整合測試較少。在最上層，您會有少量完整的端對端測試與 UAT 案例。

此結構存在是因為不同測試類型有不同的目的：

測試類型	目標	Scope	速度
單元測試	確認各項功能是否正常運作	單一函式或類別	速度快（毫秒級）
整合測試	驗證元件能一起運作	多重組成部分	中等 (秒)
端對端測試	確認完整的工作流程是否產生預期成果	完整管線	較慢（分鐘）
UAT	確保解決方案符合業務需求	商務案例	不定

從單元測試開始，你能建立對程式碼基本建構元素的信心。整合測試則確認這些區塊是否正確連接。端對端測試驗證整個系統是否能產生正確的結果。 UAT 確保利害關係人在正式部署前批准解決方案。

用 pytest 實作單元測試

單元測試著重於單獨測試 個別功能。在 Azure Databricks 中， pytest 框架提供了強大的方法來撰寫並執行 Python 程式碼的單元測試。

考慮一個資料轉換函數，依國家/地區篩選紀錄：

def filter_country_region(df, country_region="USA"):
    return df[df.iso_code == country_region]

要測試此函式，請建立一個遵循 pytest 命名慣例的測試檔案。檔案應以以下字母開頭test_或結尾：_test.py

import pytest
import pandas as pd
from transforms import filter_country_region

@pytest.fixture
def sample_data():
    """Create test data that mimics production structure."""
    return pd.DataFrame({
        'iso_code': ['USA', 'USA', 'CAN', 'GBR'],
        'value': [100, 200, 150, 175]
    })

def test_filter_country_region_default(sample_data):
    result = filter_country_region(sample_data)
    assert len(result) == 2
    assert all(result.iso_code == 'USA')

def test_filter_country_region_specific(sample_data):
    result = filter_country_region(sample_data, country_region='CAN')
    assert len(result) == 1
    assert result.iloc[0]['value'] == 150

裝飾者會 @pytest.fixture 產生 可重複使用的測試資料。此方法透過使用與實際資料結構鏡像的 合成資料集 ，保護生產資料，且不會暴露敏感資訊。

要在 Databricks 筆記本中執行這些測試，請安裝 pytest 並執行：

%pip install pytest

import pytest
retcode = pytest.main([".", "-v", "-p", "no:cacheprovider"])
assert retcode == 0, "Tests failed. Check the output above."

小提示

設計函式以回傳可預測的單一類型輸出。一個會回傳資料框或 False 的函數會變得難以測試。相反地，讓它總是回傳一個 DataFrame，即使是空的。

設計整合測試

整合測試會驗證 多個元件是否正確協同運作 。在資料管線中，這些測試確認資料在擷取、轉換與儲存階段間的正常流動。

與使用模擬資料的單元測試不同，整合測試通常針對 實際的 Databricks 資源執行。你可以從測試表讀取，套用轉換，並驗證輸出格式：

def test_pipeline_integration(spark):
    """Test that transformation pipeline produces expected schema."""
    # Read from test table
    input_df = spark.sql("SELECT * FROM test_catalog.test_schema.raw_data")
    
    # Apply transformation pipeline
    result_df = transform_pipeline(input_df)
    
    # Verify output structure
    expected_columns = ['id', 'processed_date', 'category', 'amount']
    assert result_df.columns == expected_columns
    
    # Verify data types
    assert result_df.schema['amount'].dataType.simpleString() == 'decimal(10,2)'

整合測試比單元測試需要更多的設定。建立 專門的測試架構或目錄 ，包含具代表性的資料樣本。這種隔離性防止測試影響生產資料，同時驗證真實元件互動。

這很重要

切勿對生產資料表執行整合測試。建立獨立的測試環境，資料與生產結構相似，但不包含敏感資訊。

建立端對端測試

端到端測試模擬從頭到尾 的完整工作流程 。這些測試驗證了整個流程在給予特定輸入時是否能產生預期結果。

設計端對端測試以涵蓋 真實情境：

def test_daily_processing_pipeline():
    """Validate complete daily data processing workflow."""
    # Setup: Create test input files
    test_date = "2024-01-15"
    setup_test_input_files(test_date)
    
    # Execute: Run the complete pipeline
    run_daily_pipeline(test_date)
    
    # Verify: Check final output table
    result = spark.sql(f"""
        SELECT COUNT(*) as row_count, 
               SUM(amount) as total_amount
        FROM production.daily_summary
        WHERE process_date = '{test_date}'
    """)
    
    row = result.first()
    assert row.row_count == 1000, f"Expected 1000 rows, got {row.row_count}"
    assert abs(row.total_amount - 50000.00) < 0.01
    
    # Cleanup: Remove test data
    cleanup_test_data(test_date)

端對端測試的執行時間比單元測試或整合測試更長。安排它們在 非尖峰時段 執行，或作為部署流程的一部分，而不是在每次程式碼變更時執行。

計劃進行使用者驗收測試

使用者接受測試（UAT）涉及 利害關係人驗證 您的解決方案是否符合 業務需求。過去的測試類型著重於技術正確性，UAT 則確認該解決方案能帶來商業價值。

有效的UAT需要謹慎規劃：

在開發開始前，與利害關係人定義接受標準
打造一個與製作相符的舞台環境
準備反映實際商業應用情境的測試情境
記錄每種情境的預期結果
建立回報問題的回饋流程

UAT 通常在暫 存環境中 運行，讓業務用戶可以利用真實資料與解決方案互動。考慮建立筆記本，讓利害關係人能執行以驗證特定情境：

# UAT Scenario: Monthly revenue reconciliation
# Expected outcome: Total matches finance system within 1%

finance_total = 1_250_000.00  # From finance system

pipeline_result = spark.sql("""
    SELECT SUM(revenue) as total_revenue
    FROM reporting.monthly_revenue
    WHERE month = '2024-01'
""").first()

variance = abs(pipeline_result.total_revenue - finance_total) / finance_total
print(f"Variance: {variance:.2%}")

if variance < 0.01:
    print("✓ UAT PASSED: Revenue totals match within tolerance")
else:
    print("✗ UAT FAILED: Revenue variance exceeds 1% threshold")

組織你的測驗結構

組織良好的測試結構能讓測試更易於維護與執行。請遵循以下 Azure Databricks 專案的慣例：

project/
├── src/
│   └── transforms.py
├── tests/
│   ├── unit/
│   │   └── test_transforms.py
│   ├── integration/
│   │   └── test_pipeline.py
│   └── e2e/
│       └── test_daily_workflow.py
├── notebooks/
│   └── run_tests.py
└── requirements.txt

Databricks 建議在 Python 專案中將函式及其單元測試存放 在筆記本之外 。此方法能改善 程式碼重用 ，並使測試更為直接。把函式存放在 .py Git 資料夾的檔案中，並視需要匯入筆記本。

執行測試時，請建立專用筆記本執行 pytest：

%pip install -r ../requirements.txt

import pytest
import sys

# Prevent pytest from caching to readonly filesystem
sys.dont_write_bytecode = True

# Run all tests with verbose output
retcode = pytest.main([
    "tests/",
    "-v",
    "-p", "no:cacheprovider"
])

assert retcode == 0, "Test suite failed"

在將更改部署到生產環境之前，透過建立一個 Databricks 工作來運行您的測試筆記本，以自動化測試執行。此自動化確保程式碼變更不會引發迴歸。

意見反應

此頁面對您有幫助嗎？