共用方式為

教學課程：使用 SemPy 和絕佳的期望來驗證數據 （GX）

  • 發行項

在本教學課程中，您將瞭解如何搭配使用 SemPy 與 Great Expectations （GX） 在 Power BI 語意模型上執行數據驗證。

本教學課程會示範如何：

  • 使用絕佳的預期網狀架構數據源來驗證 Fabric 工作區中數據集的條件約束（以語意連結為基礎）。
    • 設定 GX 資料內容、數據資產和期望。
    • 使用 GX 檢查點檢視驗證結果。
  • 使用語意連結來分析原始數據。

必要條件

  • 從左側瀏覽窗格中選取 [工作區 ]，以尋找並選取您的工作區。 此工作區會變成您目前的工作區。
  • 下載零售分析範例 PBIX.pbix 檔案。
  • 在您的工作區中，使用 [上傳] 按鈕將零售分析範例 PBIX.pbix 檔案上傳至工作區。

在筆記本中跟著

great_expectations_tutorial.ipynb 是本教學課程隨附的筆記本。

若要開啟本教學課程隨附的筆記本，請遵循準備系統以進行數據科學教學課程中的指示，將筆記本匯入您的工作區。

如果您想要複製並貼上此頁面中的程式碼，您可以 建立新的筆記本。

開始執行程序代碼之前，請務必將 Lakehouse 附加至筆記本 。

設定筆記本

在本節中，您會使用必要的模組和數據來設定筆記本環境。

  1. 使用%pip筆記本內的內嵌安裝功能，從 PyPI 安裝和SemPy相關的Great Expectations連結庫。
# install libraries
%pip install semantic-link great-expectations great_expectations_experimental great_expectations_zipcode_expectations

# load %%dax cell magic
%load_ext sempy
  1. 執行稍後所需模組的必要匯入：
import great_expectations as gx
from great_expectations.expectations.expectation import ExpectationConfiguration
from great_expectations_zipcode_expectations.expectations import expect_column_values_to_be_valid_zip5

設定 GX 資料內容和數據來源

若要開始使用「絕佳的期望」，您必須先設定 GX 數據內容。 內容可作為 GX 作業的進入點，並保留所有相關的組態。

context = gx.get_context()

您現在可以將網狀架構數據集新增至此內容作為 數據源 ，開始與數據互動。 本教學課程使用標準 Power BI 範例語意模型 零售分析範例 .pbix 檔案。

ds = context.sources.add_fabric_powerbi("Retail Analysis Data Source", dataset="Retail Analysis Sample PBIX")

指定數據資產

定義 數據資產 ，以指定您想要處理的數據子集。 資產可以像完整數據表一樣簡單，也可以像自定義數據分析表達式 （DAX） 查詢一樣複雜。

在這裡，您將新增多個資產：

  • Power BI 數據表
  • Power BI 量值
  • 自訂 DAX 查詢
  • 動態管理檢視 （DMV） 查詢

Power BI 數據表

將 Power BI 數據表新增為數據資產。

ds.add_powerbi_table_asset("Store Asset", table="Store")

Power BI 量值

如果您的數據集包含預先設定的量值，您可以在 SemPy 的 evaluate_measure類似 API 之後，將量值新增為資產。

ds.add_powerbi_measure_asset(
    "Total Units Asset",
    measure="TotalUnits",
    groupby_columns=["Time[FiscalYear]", "Time[FiscalMonth]"]
)

DAX

如果您想要定義自己的量值或更能控制特定資料列，您可以使用自定義 DAX 查詢來新增 DAX 資產。 在這裡，我們會將兩個 Total Units Ratio 現有的量值除以定義量值。

ds.add_powerbi_dax_asset(
    "Total Units YoY Asset",
    dax_string=
    """
    EVALUATE SUMMARIZECOLUMNS(
        'Time'[FiscalYear],
        'Time'[FiscalMonth],
        "Total Units Ratio", DIVIDE([Total Units This Year], [Total Units Last Year])
    )    
    """
)

DMV 查詢

在某些情況下，使用 動態管理檢視 （DMV） 計算作為數據驗證程式的一部分可能會很有説明。 例如，您可以追蹤資料集內的引用完整性違規數目。 如需詳細資訊，請參閱 清除資料 = 更快速的報告。

ds.add_powerbi_dax_asset(
    "Referential Integrity Violation",
    dax_string=
    """
    SELECT
        [Database_name],
        [Dimension_Name],
        [RIVIOLATION_COUNT]
    FROM $SYSTEM.DISCOVER_STORAGE_TABLES
    """
)

預期結果

若要將特定條件約束新增至資產，您必須先設定 期望套件。 將個別 期望 新增至每個套件之後，您就可以更新從新套件開始設定的數據內容。 如需可用預期的完整清單，請參閱 GX 預期資源庫。

首先，新增具有兩個期望的「零售商店套件」：

  • 有效的郵遞區號
  • 數據列計數介於80到200之間的數據表
suite_store = context.add_expectation_suite("Retail Store Suite")

suite_store.add_expectation(ExpectationConfiguration("expect_column_values_to_be_valid_zip5", { "column": "PostalCode" }))
suite_store.add_expectation(ExpectationConfiguration("expect_table_row_count_to_be_between", { "min_value": 80, "max_value": 200 }))

context.add_or_update_expectation_suite(expectation_suite=suite_store)

TotalUnits 措施

新增具有一個期望的「零售量值套件」：

  • 數據行值應大於 50,000
suite_measure = context.add_expectation_suite("Retail Measure Suite")
suite_measure.add_expectation(ExpectationConfiguration(
    "expect_column_values_to_be_between", 
    {
        "column": "TotalUnits",
        "min_value": 50000
    }
))

context.add_or_update_expectation_suite(expectation_suite=suite_measure)

Total Units Ratio DAX

新增具有一個期望的「零售 DAX 套件」：

  • 總單位比率的數據行值應介於 0.8 和 1.5 之間
suite_dax = context.add_expectation_suite("Retail DAX Suite")
suite_dax.add_expectation(ExpectationConfiguration(
    "expect_column_values_to_be_between", 
    {
        "column": "[Total Units Ratio]",
        "min_value": 0.8,
        "max_value": 1.5
    }
))

context.add_or_update_expectation_suite(expectation_suite=suite_dax)

參考完整性違規 （DMV）

新增具有一個期望的「零售 DMV 套件」：

  • RIVIOLATION_COUNT應該是 0
suite_dmv = context.add_expectation_suite("Retail DMV Suite")
# There should be no RI violations
suite_dmv.add_expectation(ExpectationConfiguration(
    "expect_column_values_to_be_in_set", 
    {
        "column": "RIVIOLATION_COUNT",
        "value_set": [0]
    }
))
context.add_or_update_expectation_suite(expectation_suite=suite_dmv)

驗證

若要實際對數據執行指定的預期，請先建立 檢查點 並將它新增至內容。 如需檢查點設定的詳細資訊，請參閱 數據驗證工作流程。

checkpoint_config = {
    "name": f"Retail Analysis Checkpoint",
    "validations": [
        {
            "expectation_suite_name": "Retail Store Suite",
            "batch_request": {
                "datasource_name": "Retail Analysis Data Source",
                "data_asset_name": "Store Asset",
            },
        },
        {
            "expectation_suite_name": "Retail Measure Suite",
            "batch_request": {
                "datasource_name": "Retail Analysis Data Source",
                "data_asset_name": "Total Units Asset",
            },
        },
        {
            "expectation_suite_name": "Retail DAX Suite",
            "batch_request": {
                "datasource_name": "Retail Analysis Data Source",
                "data_asset_name": "Total Units YoY Asset",
            },
        },
        {
            "expectation_suite_name": "Retail DMV Suite",
            "batch_request": {
                "datasource_name": "Retail Analysis Data Source",
                "data_asset_name": "Referential Integrity Violation",
            },
        },
    ],
}
checkpoint = context.add_checkpoint(
    **checkpoint_config
)

現在，請執行檢查點並將結果擷取為 pandas DataFrame，以便進行簡單的格式設定。

result = checkpoint.run()

處理並列印您的結果。

import pandas as pd

data = []

for run_result in result.run_results:
    for validation_result in result.run_results[run_result]["validation_result"]["results"]:
        row = {
            "Batch ID": run_result.batch_identifier,
            "type": validation_result.expectation_config.expectation_type,
            "success": validation_result.success
        }

        row.update(dict(validation_result.result))
        
        data.append(row)

result_df = pd.DataFrame.from_records(data)    

result_df[["Batch ID", "type", "success", "element_count", "unexpected_count", "partial_unexpected_list"]]

數據表顯示驗證結果。

從這些結果中，您可以看到所有預期都通過驗證，但透過自定義 DAX 查詢定義的「總單位 YoY 資產」除外。

診斷

使用語意連結，您可以擷取源數據，以瞭解哪些確切年份超出範圍。 語意連結提供執行 DAX 查詢的內嵌魔術。 使用語意連結來執行您傳入 GX 數據資產的相同查詢，並將產生的值可視化。

%%dax "Retail Analysis Sample PBIX"

EVALUATE SUMMARIZECOLUMNS(
    'Time'[FiscalYear],
    'Time'[FiscalMonth],
    "Total Units Ratio", DIVIDE([Total Units This Year], [Total Units Last Year])
)

數據表顯示 DAX 查詢摘要的結果。

將這些結果儲存在 DataFrame 中。

df = _

繪製結果。

import matplotlib.pyplot as plt

df["Total Units % Change YoY"] = (df["[Total Units Ratio]"] - 1)

df.set_index(["Time[FiscalYear]", "Time[FiscalMonth]"]).plot.bar(y="Total Units % Change YoY")

plt.axhline(0)

plt.axhline(-0.2, color="red", linestyle="dotted")
plt.axhline( 0.5, color="red", linestyle="dotted")

None

繪圖會顯示 DAX 查詢摘要的結果。

從繪圖中，您可以看到4月和7月稍微超出範圍，然後可以採取進一步的步驟進行調查。

儲存 GX 組態

隨著數據集中的數據隨著時間變更，您可能想要重新執行您剛執行的 GX 驗證。 目前，數據內容（包含連接的數據資產、期望套件和檢查點）暫時存回，但可以轉換成檔案內容以供日後使用。 或者，您可以具現化檔案內容（請參閱 具現化數據內容）。

context = context.convert_to_file_context()

現在您已儲存內容，請將目錄複製到 gx 您的 Lakehouse。

重要

此儲存格假設您已 將 Lakehouse 新增至筆記本。 如果沒有附加 Lakehouse，您就不會看到錯誤，但您稍後也無法取得內容。 如果您現在新增 Lakehouse，核心將會重新啟動，因此您必須重新執行整個筆記本，才能回到這一點。

# copy GX directory to attached lakehouse
!cp -r gx/ /lakehouse/default/Files/gx

現在，可以使用 建立 context = gx.get_context(project_root_dir="<your path here>") 未來的內容，以使用本教學課程中的所有設定。

例如，在新筆記本中，附加相同的 Lakehouse，並使用 context = gx.get_context(project_root_dir="/lakehouse/default/Files/gx") 來擷取內容。

相關內容

查看語意連結 /SemPy 的其他教學課程：