在現代數據工程中,確保數據模型的準確性和可靠性至關重要。SQLMesh 提供了一套強大的測試工具,用于驗證數據模型的輸出是否符合預期。本文將深入探討 SQLMesh 的測試功能,包括如何創建測試、支持的數據格式以及如何運行和調試測試。
SQLMesh 測試概述
SQLMesh 的測試功能旨在通過持續驗證每個模型的輸出來保護項目免受回歸影響。與軟件開發的單元測試類似,SQLMesh 使用預定義的輸入評估模型的邏輯,并將其輸出與每個測試提供的預期結果進行比較。這種測試方法不僅可以在每次新計劃創建時自動執行,還可以作為 CI/CD 流程的一部分按需執行。
創建測試
在 SQLMesh 中,測試套件是一個包含在項目 tests/
文件夾中的 YAML 文件,文件名以 test
開頭并以 .yaml
或 .yml
結尾。測試套件可以包含一個或多個唯一命名的單元測試,每個單元測試都有一系列屬性來定義其行為。一個單元測試至少需要指定被測試的模型、上游模型的輸入值以及目標模型的查詢和/或公共表表達式的預期輸出。
SQLMesh 支持多種方式來定義單元測試中的輸入和輸出數據:
- YAML 字典:列映射到它們的值。
- CSV:逗號分隔的值。
- SQL 查詢:針對測試連接執行 SQL 查詢以生成數據。
入門實例
在本例中,我們將使用sqlmesh_example.Full_model模型,作為sqlmesh init命令的一部分提供,定義如下:
MODEL (name sqlmesh_example.full_model,kind FULL,cron '@daily',grain item_id,audits (assert_positive_order_ids),
);SELECTitem_id,COUNT(DISTINCT id) AS num_orders,
FROMsqlmesh_example.incremental_model
GROUP BY item_id
此模型從上游的 sqlmesh_example.incremental_model 中聚合每個 item_id 的訂單數量。測試此模型的一種方法如下所示:
test_example_full_model:model: sqlmesh_example.full_modelinputs:sqlmesh_example.incremental_model:rows:- id: 1item_id: 1- id: 2item_id: 1- id: 3item_id: 2outputs:query:rows:- item_id: 1num_orders: 2- item_id: 2num_orders: 1
此測試驗證 sqlmesh_example.full_model
是否能正確統計每個 item_id
的訂單數量。它向 sqlmesh_example.incremental_model
提供三行輸入,并期望目標模型的查詢輸出兩行結果。
運行和調試測試
SQLMesh 的測試可以通過 CLI 或 Jupyter 筆記本按需執行。CLI 命令 sqlmesh test
可以用來執行所有測試,而 %run_test
筆記本魔法命令則允許在筆記本環境中執行測試。如果遇到測試失敗,可以使用 --preserve-fixtures
選項保留輸入夾具,以便進行調試。
$ sqlmesh test
.
----------------------------------------------------------------------
Ran 1 test in 0.005sOK
要運行特定的模型測試,請傳入測試套件文件名后跟 :: 和測試名稱:
$ sqlmesh test tests/test_full_model.yaml::test_example_full_model
您還可以使用通配符路徑擴展語法運行匹配模式或子字符串的測試:
$ sqlmesh test tests/test_*
測試用例實戰
CTE測試
模型查詢中的各個公用表表達式(CTE)也可以進行測試。為了演示這一點,讓我們對 sqlmesh_example.full_model
的查詢稍作修改,添加一個名為 filtered_orders_cte
的 CTE:
WITH filtered_orders_cte AS (SELECTid,item_idFROMsqlmesh_example.incremental_modelWHEREitem_id = 1
)
SELECTitem_id,COUNT(DISTINCT id) AS num_orders,
FROMfiltered_orders_cte
GROUP BY item_id
下面的測試將在聚合發生之前驗證該CTE的輸出:
test_example_full_model:model: sqlmesh_example.full_modelinputs:sqlmesh_example.incremental_model:rows:- id: 1item_id: 1- id: 2item_id: 1- id: 3item_id: 2outputs:ctes:filtered_orders_cte:rows:- id: 1item_id: 1- id: 2item_id: 1query:rows:- item_id: 1num_orders: 2
csv文件
這就是我們如何定義與第一個示例相同的測試,但輸入數據格式為CSV:
test_example_full_model:model: sqlmesh_example.full_modelinputs:sqlmesh_example.incremental_model:format: csvrows: |id,item_id1,12,13,2outputs:query:rows:- item_id: 1num_orders: 2- item_id: 2num_orders: 1
sql查詢
這就是我們如何能夠將上述第一個示例中的相同測試定義為這樣一種形式,只不過輸入數據是通過 SQL 查詢生成的:
test_example_full_model:model: sqlmesh_example.full_modelinputs:sqlmesh_example.incremental_model:query: |SELECT 1 AS id, 1 AS item_idUNION ALLSELECT 2 AS id, 1 AS item_idUNION ALLSELECT 3 AS id, 2 AS item_idoutputs:query:rows:- item_id: 1num_orders: 2- item_id: 2num_orders: 1
數據文件
SQLMesh支持從外部文件加載數據。要實現這一點,你可以使用pathattribute,它指定要加載的數據的路徑名:
test_example_full_model:model: sqlmesh_example.full_modelinputs:sqlmesh_example.incremental_model:format: csvpath: filepath/test_data.csv
如果省略format,則該文件將作為YAML文檔加載。
省略列
對于寬表(即具有眾多列的表),定義完整的輸入和預期輸出可能會變得繁瑣。因此,如果某些列可以安全地忽略,那么它們可以從任何行中省略,并且對于該行,其值將被視為 NULL。
此外,可以通過將 partial 設置為 true 來僅測試感興趣的輸出列的一部分:
outputs:query:partial: truerows:- <column_name>: <column_value>...
當缺失的列不能被視為 NULL 值,但我們仍希望忽略這些列時,此設置非常有用。若要將此設置應用于所有預期輸出,請在“輸出”鍵下進行設置:
outputs:partial: true...
凍結時間
某些模型可能會使用計算給定時間點 datetime 值的 SQL 表達式,例如 CURRENT_TIMESTAMP。由于這些表達式是非確定性的,僅僅指定預期的輸出值不足以對其進行測試。
通過設置 execution_time 宏變量來模擬測試上下文中的當前時間解決了這個問題,從而使其值具有確定性。
以下示例展示了如何使用 execution_time 來測試使用 CURRENT_TIMESTAMP 計算的列。我們將要測試的模型定義如下:
MODEL (name colors,kind FULL
);SELECT'Yellow' AS color,CURRENT_TIMESTAMP AS created_at
測試文件如下:
test_colors:model: colorsoutputs:query:- color: "Yellow"created_at: "2023-01-01 12:05:03"vars:execution_time: "2023-01-01 12:05:03"
還可以為執行時間設置時區,方法是在時間戳字符串中包含該時區。
如果提供了時區,目前的要求是測試的預期日期時間值必須是無時區的時戳,這意味著它們需要相應地進行偏移。
如果我們希望將時間凍結為 UTC+2,以下是上述測試的編寫方式:
test_colors:model: colorsoutputs:query:- color: "Yellow"created_at: "2023-01-01 10:05:03"vars:execution_time: "2023-01-01 12:05:03+02:00"
自動生成測試
手動創建測試可能會顯得單調乏味且容易出錯,這就是為什么 SQLMesh 還提供了使用 create_test 命令來實現自動化處理這一過程的方法。
此命令能夠為給定的模型生成完整的測試,只要其上游模型的表存在于項目的數據倉庫中,并且這些表中已有數據即可。
實戰示例
在這個示例中,我們將展示如何為 sqlmesh_example.incremental_model 生成測試用例。sqlmesh_example.incremental_model 是作為 sqlmesh init 命令的一部分提供的另一個模型,其定義如下:
MODEL (name sqlmesh_example.incremental_model,kind INCREMENTAL_BY_TIME_RANGE (time_column event_date),start '2020-01-01',cron '@daily',grain (id, event_date)
);SELECTid,item_id,event_date,
FROMsqlmesh_example.seed_model
WHEREevent_date BETWEEN @start_date AND @end_date
首先,我們需要明確上游模型 sqlmesh_example.seed_model 的輸入數據。create_test 命令的執行始于對項目的數據倉庫發出用戶自定義的查詢,以獲取這些數據。
例如,以下查詢將從與模型 sqlmesh_example.seed_model 相對應的表中返回三行數據:1
SELECT * FROM sqlmesh_example.seed_model LIMIT 3
接下來,請留意 sqlmesh_example.incremental_model 中包含一個引用了 @start_date 和 @end_date 宏變量的過濾條件。為了使生成的測試具有確定性,從而確保它總是能夠成功,我們需要定義這些變量,并修改上述查詢以相應地約束 event_date。
如果我們將 @start_date 設為 ‘2020-01-01’ 并將 @end_date 設為 ‘2020-01-04’,上述查詢需要修改為:
SELECT * FROM sqlmesh_example.seed_model WHERE event_date BETWEEN '2020-01-01' AND '2020-01-04' LIMIT 3
運行此操作會創建以下新測試,其位于 tests/test_incremental_model.yaml 文件中:
test_incremental_model:model: sqlmesh_example.incremental_modelinputs:sqlmesh_example.seed_model:- id: 1item_id: 2event_date: 2020-01-01- id: 2item_id: 1event_date: 2020-01-01- id: 3item_id: 3event_date: 2020-01-03outputs:query:- id: 1item_id: 2event_date: 2020-01-01- id: 2item_id: 1event_date: 2020-01-01- id: 3item_id: 3event_date: 2020-01-03vars:start: '2020-01-01'end: '2020-01-04'
配置測試連接
對于給定的測試,可以更改測試連接。例如,當被測試的模型無法正確編譯為默認測試引擎的方言時,這可能會很有用。
以下示例通過修改 test_example_full_model 來演示這一點,使其針對單線程本地 Spark 進程運行,該進程在項目的 config.yaml 文件中的 spark_testing 網關中定義為 test_connection:
gateways:local:connection:type: duckdbdatabase: db.dbspark_testing:test_connection:type: sparkconfig:# Run Spark locally with one worker thread"spark.master": "local"# Move data under /tmp so that it is only temporarily persisted"spark.sql.warehouse.dir": "/tmp/data_dir""spark.driver.extraJavaOptions": "-Dderby.system.home=/tmp/derby_dir"default_gateway: localmodel_defaults:dialect: duckdb
修改測試用例:
test_example_full_model:gateway: spark_testing# ... the other test attributes remain the same
最后總結
SQLMesh 的測試功能為數據工程師提供了一個強大的工具,用于確保數據模型的準確性和可靠性。通過自動化測試過程,SQLMesh 幫助團隊在每次模型變更后都能快速驗證其正確性。無論是手動創建測試還是使用自動測試生成工具,SQLMesh 都能有效地提升數據工程的質量和效率。對于希望在數據工程中實現更高可靠性的團隊來說,SQLMesh 是一個不可或缺的工具。