具体化された湖のビューのデータ品質

medallion アーキテクチャでは、すべての段階でデータ品質を適用する必要があります。 データ品質が低いと、間違った分析情報や運用上の非効率性が生じる可能性があります。

この記事では、Microsoft Fabric の具体化されたレイク ビュー (MLV) でデータ品質チェックを実装する方法について説明します。

データ品質を実装する

行レベルの制約は各行を単独で検証します(例:「この行のcustomerName はnullではない」)。 しかし、多くの実際の品質問題は、列全体の性質です。

• 一意性 — order_id のようなプライマリキーやビジネスキーには重複キーがありません。
• 完全性 — customer_id のようなクリティカルカラムにはヌルが含まれません。
• 統計的形状 — 列の平均、大きさ、または分布が期待範囲内に収まる。

リフレッシュ時にこれらの問題を検出することで、不正なデータがブロンズ→シルバー→ゴールドチェーンを通じて下流に伝わるのを防ぎ、毎回の実行で監査可能な品質の信号が得られます。

• 行レベルのデータ品質制約
• カラムレベルのデータ品質チェック

行レベルのデータ品質制約

Fabric のマテリアライズド レイク ビュー (MLV) では、ビューに制約を定義してデータ品質を維持します。 明示的なチェックを行わないと、小さなデータの問題によって処理時間が増加したり、パイプラインが失敗したりする可能性があります。

行が制約に違反している場合は、次のいずれかのアクションを使用できます。

• FAIL: 最初の制約違反で MLV の更新を停止します。 これは、 FAILを指定しない場合でも、既定の動作です。

• DROP: 処理を続行し、制約に違反するレコードを削除します。 系列ビューには、削除されたレコードの数が表示されます。

具体化されたレイク ビューでデータ品質チェックを定義する

具体化されたレイク ビューを作成するときに、制約 (更新中に各行を検証するデータ品質ルール) を定義できます。 制約は、すべての行が満たす必要があるブール式です。 渡された行は出力テーブルに書き込まれます。 失敗した行は、違反時の設定に従って処理されます。警告なしに削除されるか、更新全体が失敗します。

次の例では、cust_blank フィールドが null でないかどうかをチェックする制約customerNameを定義します。 制約により、null customerName の行が処理から除外されます。

CREATE OR REPLACE MATERIALIZED LAKE VIEW IF NOT EXISTS silver.customers_enriched  
(CONSTRAINT cust_blank CHECK (customerName is not null) on MISMATCH DROP)
AS
SELECT
    c.customerID,
    c.customerName,
    c.contact, 
    CASE  
       WHEN COUNT(o.orderID) OVER (PARTITION BY c.customerID) > 0 THEN TRUE  
       ELSE FALSE  
    END AS has_orders 
FROM bronze.customers c LEFT JOIN bronze.orders o 
ON c.customerID = o.customerID;

システム内蔵関数

UPPER()、LOWER()、TRIM()、COALESCE()、INITCAP()、DATE_FORMAT() などの組み込みの Spark/SQL 関数は、CREATE と LINEAGE の両方の更新のすべての MLV コンテキストで完全にサポートされています。

CREATE MATERIALIZED LAKE VIEW sample_lakehouse.silver.names (
CONSTRAINT substring_check
CHECK (SUBSTRING(name, 1, 2) = 'Al') ON MISMATCH drop
) AS
SELECT id, name
FROM (VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Ann')) AS t(id, name)

UDF – 同じノートブックで定義・登録

同じノートブック内の spark.udf.register() に登録された UDF は、すべてのコンテキストで CREATE でサポートされます。 LINEAGE の更新では、スケジュールされたノートブックの実行の一部として UDF 定義が実行されるため、PySpark コンテキストのみがサポートされます。

from pyspark.sql import SparkSession
from pyspark.sql.types import StringType
from pyspark.sql.functions import expr

spark = SparkSession.builder.getOrCreate()

# UDF: Extract domain from email
def extract_email_domain(email):
    if email is None or '@' not in email:
        return None
    return email.split('@')[1]

# Registration
spark.udf.register(
    "udf_email_domain",
    extract_email_domain,
    StringType()
)

@fmlv.materialized_lake_view(
    name="udf_testing_silver.mlv_high_value_customers",
    comment="High-value customers identified by UDF criteria",
    table_properties={"delta.enableChangeDataFeed": "true"}
)
def mlv_high_value_customers():
    return spark.sql("""
        SELECT 
            c.customer_id,
            c.name,
            c.email,
            udf_email_domain(c.email) as email_domain,
            c.segment,
            c.lifetime_value,
            total_transactions.total_amount,
            total_transactions.txn_count
        FROM udf_testing_bronze.customers c
        INNER JOIN (
            SELECT 
                customer_id,
                SUM(amount) as total_amount,
                COUNT(*) as txn_count
            FROM udf_testing_bronze.transactions
            WHERE udf_is_positive(amount)
            GROUP BY customer_id
            HAVING SUM(amount) > 1000
        ) total_transactions ON c.customer_id = total_transactions.customer_id
        WHERE udf_validate_customer(c.email, c.age)
            AND c.segment IN ('premium', 'vip')
    """)

print("✓ Created mlv_high_value_customers")

サードパーティライブラリ - Pandas UDFs

Pandas UDF などのサード パーティ製ライブラリを使用すると、ベクター化された処理を使用してデータ品質ルールを実装できます。 組み込み関数では不可能なカスタム ビジネス ロジック、統計チェック、パターン検出などの高度な検証が可能になります。 これは、MLV の作成と更新中にスケーラブルで再利用可能なデータ品質制約を構築するのに役立ちます。

  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
  from datetime import datetime

  import pandas as pd
  from pyspark.sql.types import BooleanType
  def pandas_check_impl(val):
    # Reject if value < median of [100, 200, 300]
    return val >= pd.Series([100, 200, 300]).median()
  spark.udf.register("pandas_check", pandas_check_impl, BooleanType())

  @fmlv.materialized_lake_view(
    name="silver.pyspark_from_two_sqlmlv_inner_pandas",
    comment="PySpark MLV INNER JOIN using pandas-based constraint and DROP violations"
  )
  @fmlv.check(
    name="dq_pandas_check",
    condition="pandas_check(l3)",
    action="DROP"
  )
  def pyspark_from_two_sqlmlv_inner_pandas():
    # Define the function

    # Register the function as a Spark UDF

    # Read source tables
    df1 = spark.table("silver.base_sqlmlv")
    df2 = spark.table("silver.base_sqlmlv")

    # Rename columns for unique join
    df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
    df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])

    # Perform INNER JOIN
    df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
    return df


  df = spark.table("silver.pyspark_from_two_sqlmlv_inner_pandas")
  # All amounts should be >= median (200)
  assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
  print("PySpark MLV INNER JOIN with pandas-based DQ DROP passed")

カスタムライブラリ – Python wheel (.whl)

jar ファイルまたはホイール ファイルとしてパッケージ化された関数は、(環境設定を使用して) Fabric クラスターにインストールし、MLV 定義で使用できます。 PYSpark コンテキストでは、CREATE と LINEAGE REFRESH がサポートされています。 詳細については、 Fabric 環境でのカスタム ライブラリの管理 に関するページを参照してください。

  %%pyspark
  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, TimestampType, StringType, DoubleType, BooleanType
  from datetime import datetime

  from pyspark.sql.types import BooleanType
  from custom_dq_lib import threshold_check
  def custom_check_impl(val):
    return threshold_check(val, threshold=200)
  spark.udf.register("custom_check", custom_check_impl, BooleanType())

  @fmlv.materialized_lake_view(
    name="silver.pyspark_from_two_sqlmlv_inner_custom_whl",
    comment="PySpark MLV INNER JOIN using custom DQ library and DROP violations",
    replace=True
  )
  @fmlv.check(
    name="dq_custom_check",
    condition="custom_check(l3)",
    action="DROP"
  )
  def pyspark_from_two_sqlmlv_inner_custom():
    # Wrap the custom function as Spark UDF


    # Read source tables
    df1 = spark.table("silver.base_sqlmlv")
    df2 = spark.table("silver.base_sqlmlv")

    # Rename columns for unique join
    df_left = df1.select([df1[col].alias(f"l{i+1}") for i, col in enumerate(df1.columns)])
    df_right = df2.select([df2[col].alias(f"r{i+1}") for i, col in enumerate(df2.columns)])

    # Perform INNER JOIN
    df = df_left.join(df_right, df_left.l1 == df_right.r1, "inner")
    return df


    df = spark.table("silver.pyspark_from_two_sqlmlv_inner_custom_whl")
    # All amounts should be >= threshold (200)
    assert all(df.select("l3").rdd.map(lambda r: r[0] >= 200).collect()), "Unexpected low-value rows found"
    print("PySpark MLV INNER JOIN with custom DQ library passed")

Fabricのユーザーデータ関数

Fabric ユーザー データ関数 (UDF) は、Fabric ワークスペースで一元的に定義および管理されます。 これらは、セッションごとに再登録する必要なく、任意のノートブックまたはパイプラインで使用できるため、運用 MLV パイプラインに最適です。 この機能は、CREATE と LINEAGE の両方の更新で PySpark コンテキストでのみサポートされます。 ユーザー データ関数の詳細については、こちらをご覧ください。 詳細については、「 ファブリック ユーザー データ関数の概要」を参照してください。

  %%pyspark
  import fmlv
  from pyspark.sql import functions as F
  from notebookutils import udf

  myFuncs = udf.getFunctions("UserDataFunction_1")

  def add_greeting_column(df):
    pdf = df.toPandas()
    pdf["greeting"] = pdf["name"].apply(lambda n: myFuncs.hello_fabric(n))
    return spark.createDataFrame(pdf)

  import fmlv
  from pyspark.sql.types import StructType, StructField, IntegerType, StringType, DoubleType

  # -------------------------------------------------------------
  # Define base PySpark MLV (no SQL)
  # -------------------------------------------------------------
  @fmlv.materialized_lake_view(
    name="silver.base_pysparkmlv",
    comment="Base MLV created using PySpark",
    replace=True
  )
  def base_pysparkmlv():

    schema = StructType([
        StructField("id", IntegerType(), True),
        StructField("name", StringType(), True),
        StructField("amount", DoubleType(), True),
        StructField("country", StringType(), True)
    ])

    data = [
        (1, "Alice", 100.0, "US"),
        (2, "Bob", 200.0, "UK"),
        (3, "Charlie", 300.0, "UK")
    ]

    return spark.createDataFrame(data, schema)

  @fmlv.materialized_lake_view(
    name="silver.mlv_udfn_null_test",
    replace=False
  )
  @fmlv.check(
    name="null_check",
    condition="greeting IS NOT NULL",
    action="FAIL"
  )
  def mlv_udfn_null_test():
    df = spark.table("silver.base_pysparkmlv")
    return add_greeting_column(df)

カラムレベルのデータ品質チェック

どのように機能するのか

MLVの定義とは別に、ノートブックのセルに配置する3つのPythonデコレーターでカラムレベルのデータ品質(DQ)チェックを定義します。

このスイートはMLVを名前でターゲットにしているため、データ品質をMLVの定義とは独立して管理、実行、スケジューリングが可能で、リフレッシュをトリガーせずにチェックを実行することも可能です。

ステップ1: ランタイムを設定する

Fabric ランタイム バージョン用の依存関係を設定してから、カラム レベルのデータ品質チェックを定義してください。

2.0以前のFabricランタイム

PyDeequ 1.7.0ホイールをダウンロードし、Fabric環境にカスタムライブラリとしてアップロードしてください。 手順については「カスタムライブラリのアップロード」をご覧ください。

Fabric ランタイム 2.0以降

以下のSparkプロパティを追加してください:

これらのプロパティは環境やノートブックのいずれかに追加できます:

• 環境:Spark Compute>Sparkプロパティの下にプロパティを追加し、環境を保存して公開します。 手順については「 カスタム環境へのスパーク設定の追加」をご覧ください。
• ノートブック: Spark セッションを開始する前に、最初のコードセルで conf コマンドの %%configure オブジェクトにプロパティを追加してください。 手順については、 Sparkセッション構成のmagicコマンドを参照してください。

ステップ2:カラムレベルのチェックを定義する

この例では、sales_db.daily_orders 内の mlv_name で指定された @fmlv.dq_suite マテリアライズド レイクビューに対する列レベルのデータ品質チェックを定義します。

スイート登録とそのエントリー関数を単一のデータ品質セルにまとめましょう。 以下の例は、1回の検証パスで 2つのチェック を実行します — customer_id は完全でなければならず(nullなし)、 order_id は一意でなければなりません。

# Data quality cell — separate from the MLV definition
@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
def daily_orders_dq():
    pass

# Entry function — returns a PyDeequ VerificationResult DataFrame
def completeness_and_uniqueness(df):
    from pydeequ.checks import Check, CheckLevel
    from pydeequ.verification import VerificationSuite, VerificationResult

    check = Check(spark, CheckLevel.Error, "Completeness & Uniqueness")
    result = (
        VerificationSuite(spark)
        .onData(df)
        .addCheck(
            check.isComplete("customer_id")   # check 1: no nulls
                 .isUnique("order_id")        # check 2: no duplicates
        )
        .run()
    )
    return VerificationResult.checkResultsAsDataFrame(spark, result)

実行を失敗させずにメトリクスを追加しましょう。 @fmlv.analyzeパスを積み重ねて、可観測性のためにカラムのメトリクス（サイズ、完全性、平均など）を計算します。 分析結果は常に記録され、実行をブロックすることはありません。検証ルールが進化している間は有用です。

@fmlv.dq_suite(mlv_name="sales_db.daily_orders", replace=True)
@fmlv.validate("completeness_and_uniqueness", "FAIL")
@fmlv.analyze("column_profiling")
def daily_orders_dq():
    pass

def column_profiling(df):
    from pydeequ.analyzers import AnalysisRunner, AnalyzerContext, Size, Completeness, Mean
    result = (
        AnalysisRunner(spark)
        .onData(df)
        .addAnalyzer(Size())
        .addAnalyzer(Completeness("customer_id"))
        .addAnalyzer(Mean("amount"))
        .run()
    )
    return AnalyzerContext.successMetricsAsDataFrame(spark, result)

結果をコントロールする: FAIL vs. WARN

@fmlv.validateのアクションは、チェック失敗時に何が起こるかを決定します:

• FAIL — チェックの失敗により、データ品質実行は失敗し、それに依存する下流のMLVをブロックします。 主キーや必須フィールドなどのハードコントラクトに使うのが良いでしょう。
• WARN — 失敗は記録されますが、実行は成功し、下流のMLVは ブロック されません。 ソフトルールやアドバイザリールールに使用します。

ステップ3:スケジュール内でデータ品質チェックを有効にする

列レベルのチェックがリフレッシュスケジュールの一部として実行されるかどうかを選択してください。 「新しいスケジュール作成」エクスペリエンスの詳細設定では、データ品質チェックのドロップダウンで2つのオプションがあります。

• 毎回の実行後に確認 してください — そのスケジュール内の各MLV更新が完了するたびに、データ品質スイートは自動的に動作します。
• チェックはなし — そのスケジュールではデータ品質がスキップされます。MLVは設定された周期で引き続き更新されます。

スケジュールの詳細設定にあるデータ品質チェックの設定 — 毎回の実行後にチェックするか、チェックなしを選択できます。

チェック失敗がパイプラインに与える影響

リネージビューでは、gold.daily_dq_reportusage のデータ品質チェックが FAIL で定義され、失敗状態になりました。 その結果、その下流で依存するブランチ (gold.monthly_dq_report…) は自動的に スキップされ、独立したブランチ (gold.expected_regions、gold.region_mlv_dq_rep…) は通常どおり完了します。

FAILのデータ品質チェックがチェーンを停止させます。失敗したノードの下流依存ノードはスキップされますが、独立した枝は継続されます。

実行順序。 スイートに両方の型がある場合、すべての @fmlv.validate 関数が最初に(定義順に)、次にすべての @fmlv.analyze 関数が実行されます。 失敗したチェックは記録され、実行は続行されます。すべてのチェックが完了した後に全体の結果が決まります。実行は FAILアクションチェックが失敗した場合のみ失敗します。

ステップ4:データ品質レポートの閲覧と分析

実行履歴は、各実行でデータ品質がオンかオフかを反映します。 また、リフレッシュに関係なく、ノートブックからいつでもスイートをオンデマンドで実行することも可能です。

系譜ビューで任意のMLVノードを選択し、詳細パネルの 「データ品質 」タブを開くと、そのランの結果が表示されます。

• 計算されたメトリクス — Deequアナライザーによって計算される列レベルの集計指標（例: regionのCompleteness）。
• 検証結果 — 総チェック数、合格数、失敗数の合計をまとめ、各制約とその合否 ステータスが続き、どのルールが合格・不合格かを正確に確認できます。

データ品質タブでは、選択したMLV実行の計算された指標と制約ごとの検証結果が表示されます。

MLVの実行履歴には各実行のデータ品質状態( 成功、 失敗、スイートが実行されていない場合の表示なし)も表示されるため、各ノードを開かなくても問題のある実行を一目で見つけられます。

対応チェックとアナライザー

カラムレベルのDQは Amazon Deequによって実現されているため、エントリー機能内でDeequのライブラリ全体を利用できます。 検証チェック（@fmlv.validateで使用）は、合格/不合格の結果を生成するCheck とともに使用される制約です。アナライザー（@fmlv.analyze とともに使用）は、合格/不合格のないメトリクスを計算します。 最も一般的な例は以下の表に示されています。

検証チェック — @fmlv.validate

アナライザー/指標 — @fmlv.analyze

オープンソースの Amazon Deequ ライブラリは、これらのチェックと分析ツールを提供しています。 完全かつ権威あるリストについては、 github.com/awslabs/deequ のDeequプロジェクトをご覧ください。

Summary

1. MLVを作成し 、その完全限定名を記録してください。
2. 別のデータ品質セルで、@fmlv.dq_suite(mlv_name=…)を使用してスイートを登録します。
3. チェックを追加 — @fmlv.validate("<fn>", "FAIL"|"WARN") および/または @fmlv.analyze("<fn>") をスタックし、PyDeequ DataFramesを返す対応するエントリ関数を書く。
4. ノートブックからオンデマンドで実行するか、スケジュール作成時にデータ品質の切り替えをオンにしてください。
5. 結果を系譜のデータ品質タブと実行履歴で確認してください。

関連コンテンツ

• Materialized Lake Viewsのデータ品質 — Microsoft Learn
• Materialized Lake Viewsの概要 — Microsoft Learn
• DeeQuのドキュメント

関連コンテンツ

• マテリアライズド レイク ビューを更新する
• データ品質に関する Power BI レポート