期待値の推奨事項と高度なパターン

この記事には、エクスペクティブを大規模に実装するための推奨事項と、エクスペクティブによってサポートされる高度なパターンの例が含まれています。 これらのパターンでは、複数のデータセットを期待値と組み合わせて使用し、ユーザーは具体化されたビュー、ストリーミング テーブル、および期待値の構文とセマンティクスを理解する必要があります。

期待値の動作と構文の基本的な概要については、「 パイプラインの期待値を使用してデータ品質を管理する」を参照してください。

ポータブルで再利用可能な期待

Databricks では、移植性を向上させ、メンテナンスの負担を軽減するための期待を実装する際に、次のベスト プラクティスを推奨しています。

推奨事項

インパクト

期待定義をパイプライン ロジックとは別に格納します。

複数のデータセットやパイプラインに期待事項を簡単に適用できます。 パイプラインのソースコードを変更せずに、期待値を更新、監査、維持します。

カスタムタグを追加して、関連する期待値のグループを作成します。

タグに基づいて期待をフィルタリングします。

類似したデータセットに一貫して期待値を適用します。

複数のデータセットとパイプラインで同じ期待値を使用して、同一のロジックを評価します。

次の例は、Delta テーブルまたはディクショナリを使用して中央の期待値リポジトリを作成する方法を示しています。 次に、カスタム Python 関数がこれらの期待値をサンプル パイプラインのデータセットに適用します。

次の例では、ルールを維持するためにrulesという名前のテーブルを作成します：

CREATE OR REPLACE TABLE
  rules
AS SELECT
  col1 AS name,
  col2 AS constraint,
  col3 AS tag
FROM (
  VALUES
  ("website_not_null","Website IS NOT NULL","validity"),
  ("fresh_data","to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'","maintained"),
  ("social_media_access","NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)","maintained")
)

次の Python の例では、 rules テーブルのルールに基づいてデータ品質の期待値を定義しています。 get_rules() 関数は、rules テーブルからルールを読み取り、関数に渡された tag 引数に一致するルールを含む Python ディクショナリを返します。

この例では、 @dlt.expect_all_or_drop() デコレーターを使用してディクショナリを適用し、データ品質制約を適用します。

たとえば、 validity でタグ付けされたルールに失敗したレコードは、 raw_farmers_market テーブルから削除されます。

import dlt
from pyspark.sql.functions import expr, col

def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  df = spark.read.table("rules").filter(col("tag") == tag).collect()
  return {
      row['name']: row['constraint']
      for row in df
  }

@dlt.table
@dlt.expect_all_or_drop(get_rules('validity'))
def raw_farmers_market():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table
@dlt.expect_all_or_drop(get_rules('maintained'))
def organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
  )

次の例では、ルールを維持するための Python モジュールを作成します。 この例では、パイプラインのソース コードとして使用されるノートブックと同じフォルダー内の rules_module.py という名前のファイルに、このコードを格納します。

def get_rules_as_list_of_dict():
  return [
    {
      "name": "website_not_null",
      "constraint": "Website IS NOT NULL",
      "tag": "validity"
    },
    {
      "name": "fresh_data",
      "constraint": "to_date(updateTime,'M/d/yyyy h:m:s a') > '2010-01-01'",
      "tag": "maintained"
    },
    {
      "name": "social_media_access",
      "constraint": "NOT(Facebook IS NULL AND Twitter IS NULL AND Youtube IS NULL)",
      "tag": "maintained"
    }
  ]

次の Python の例では、 rules_module.py ファイルで定義されているルールに基づいて、データ品質の期待値を定義しています。 get_rules()関数は、渡された tag 引数に一致するルールを含む Python ディクショナリを返します。

この例では、 @dlt.expect_all_or_drop() デコレーターを使用してディクショナリを適用し、データ品質制約を適用します。

たとえば、 validity でタグ付けされたルールに失敗したレコードは、 raw_farmers_market テーブルから削除されます。

import dlt
from rules_module import *
from pyspark.sql.functions import expr, col


def get_rules(tag):
  """
    loads data quality rules from a table
    :param tag: tag to match
    :return: dictionary of rules that matched the tag
  """
  return {
    row['name']: row['constraint']
    for row in get_rules_as_list_of_dict()
    if row['tag'] == tag
  }

@dlt.table
@dlt.expect_all_or_drop(get_rules('validity'))
def raw_farmers_market():
  return (
    spark.read.format('csv').option("header", "true")
      .load('/databricks-datasets/data.gov/farmers_markets_geographic_data/data-001/')
  )

@dlt.table
@dlt.expect_all_or_drop(get_rules('maintained'))
def organic_farmers_market():
  return (
    dlt.read("raw_farmers_market")
      .filter(expr("Organic = 'Y'"))
  )

行数の検証

次の例では、 table_a と table_b の間の行数の等価性を検証して、変換中にデータが失われないことを確認します。

Delta Live Tables の行数検証グラフと期待使用量 
@dlt.view(
  name="count_verification",
  comment="Validates equal row counts between tables"
)
@dlt.expect_or_fail("no_rows_dropped", "a_count == b_count")
def validate_row_counts():
  return spark.sql("""
    SELECT * FROM
      (SELECT COUNT(*) AS a_count FROM LIVE.table_a),
      (SELECT COUNT(*) AS b_count FROM LIVE.table_b)""")

CREATE OR REFRESH MATERIALIZED VIEW count_verification(
  CONSTRAINT no_rows_dropped EXPECT (a_count == b_count)
) AS SELECT * FROM
  (SELECT COUNT(*) AS a_count FROM LIVE.table_a),
  (SELECT COUNT(*) AS b_count FROM LIVE.table_b)

レコードの欠落検出

次の例では、予期されるすべてのレコードがreportテーブルに存在することを検証します：

Delta Live Tables の欠落行検出グラフと期待される使用量 
@dlt.view(
  name="report_compare_tests",
  comment="Validates no records are missing after joining"
)
@dlt.expect_or_fail("no_missing_records", "r_key IS NOT NULL")
def validate_report_completeness():
  return (
    dlt.read("validation_copy").alias("v")
      .join(
        dlt.read("report").alias("r"),
        on="key",
        how="left_outer"
      )
      .select(
        "v.*",
        "r.key as r_key"
      )
  )

CREATE OR REFRESH MATERIALIZED VIEW report_compare_tests(
  CONSTRAINT no_missing_records EXPECT (r_key IS NOT NULL)
)
AS SELECT v.*, r.key as r_key FROM LIVE.validation_copy v
  LEFT OUTER JOIN LIVE.report r ON v.key = r.key

主キーの一意性

次の例では、テーブル間のプライマリ・キー制約を検証します。

Delta Live Tables の主キー一意性グラフと期待値の使用状況 
@dlt.view(
  name="report_pk_tests",
  comment="Validates primary key uniqueness"
)
@dlt.expect_or_fail("unique_pk", "num_entries = 1")
def validate_pk_uniqueness():
  return (
    dlt.read("report")
      .groupBy("pk")
      .count()
      .withColumnRenamed("count", "num_entries")
  )

CREATE OR REFRESH MATERIALIZED VIEW report_pk_tests(
  CONSTRAINT unique_pk EXPECT (num_entries = 1)
)
AS SELECT pk, count(*) as num_entries
  FROM LIVE.report
  GROUP BY pk

スキーマ進化パターン

次の例は、追加の列のスキーマ進化を処理する方法を示しています。 データソースを移行する場合や、複数のバージョンのアップストリーム データを処理する場合は、このパターンを使用して、データ品質を強化しながら下位互換性を確保します。

Delta Live Tables スキーマ進化の検証と期待値の使用 
@dlt.table
@dlt.expect_all_or_fail({
  "required_columns": "col1 IS NOT NULL AND col2 IS NOT NULL",
  "valid_col3": "CASE WHEN col3 IS NOT NULL THEN col3 > 0 ELSE TRUE END"
})
def evolving_table():
  # Legacy data (V1 schema)
  legacy_data = spark.read.table("legacy_source")

  # New data (V2 schema)
  new_data = spark.read.table("new_source")

  # Combine both sources
  return legacy_data.unionByName(new_data, allowMissingColumns=True)

CREATE OR REFRESH MATERIALIZED VIEW evolving_table(
  -- Merging multiple constraints into one as expect_all is Python-specific API
  CONSTRAINT valid_migrated_data EXPECT (
    (col1 IS NOT NULL AND col2 IS NOT NULL) AND (CASE WHEN col3 IS NOT NULL THEN col3 > 0 ELSE TRUE END)
  ) ON VIOLATION FAIL UPDATE
) AS
  SELECT * FROM new_source
  UNION
  SELECT *, NULL as col3 FROM legacy_source;

範囲ベースの検証パターン

次の例は、新しいデータポイントを過去の統計範囲に対して検証し、データ フローの外れ値と異常を特定する方法を示しています。

Delta Live Tables の範囲ベースの検証 (期待値の使用状況を使用) 
@dlt.view
def stats_validation_view():
  # Calculate statistical bounds from historical data
  bounds = spark.sql("""
    SELECT
      avg(amount) - 3 * stddev(amount) as lower_bound,
      avg(amount) + 3 * stddev(amount) as upper_bound
    FROM historical_stats
    WHERE
      date >= CURRENT_DATE() - INTERVAL 30 DAYS
  """)

  # Join with new data and apply bounds
  return spark.read.table("new_data").crossJoin(bounds)

@dlt.table
@dlt.expect_or_drop(
  "within_statistical_range",
  "amount BETWEEN lower_bound AND upper_bound"
)
def validated_amounts():
  return dlt.read("stats_validation_view")

CREATE OR REFRESH MATERIALIZED VIEW stats_validation_view AS
  WITH bounds AS (
    SELECT
    avg(amount) - 3 * stddev(amount) as lower_bound,
    avg(amount) + 3 * stddev(amount) as upper_bound
    FROM historical_stats
    WHERE date >= CURRENT_DATE() - INTERVAL 30 DAYS
  )
  SELECT
    new_data.*,
    bounds.*
  FROM new_data
  CROSS JOIN bounds;

CREATE OR REFRESH MATERIALIZED VIEW validated_amounts (
  CONSTRAINT within_statistical_range EXPECT (amount BETWEEN lower_bound AND upper_bound)
)
AS SELECT * FROM stats_validation_view;

無効なレコードを隔離する

このパターンは、期待値と一時テーブルおよびビューを組み合わせて、パイプラインの更新中にデータ品質メトリクスを追跡し、ダウンストリーム操作で有効なレコードと無効なレコードに対して別々の処理パスを有効にします。

Delta Live Tables のデータ検疫パターン (予想される使用量) 
import dlt
from pyspark.sql.functions import expr

rules = {
  "valid_pickup_zip": "(pickup_zip IS NOT NULL)",
  "valid_dropoff_zip": "(dropoff_zip IS NOT NULL)",
}
quarantine_rules = "NOT({0})".format(" AND ".join(rules.values()))

@dlt.view
def raw_trips_data():
  return spark.readStream.table("samples.nyctaxi.trips")

@dlt.table(
  temporary=True,
  partition_cols=["is_quarantined"],
)
@dlt.expect_all(rules)
def trips_data_quarantine():
  return (
    dlt.readStream("raw_trips_data").withColumn("is_quarantined", expr(quarantine_rules))
  )

@dlt.view
def valid_trips_data():
  return dlt.read("trips_data_quarantine").filter("is_quarantined=false")

@dlt.view
def invalid_trips_data():
  return dlt.read("trips_data_quarantine").filter("is_quarantined=true")

CREATE TEMPORARY STREAMING LIVE VIEW raw_trips_data AS
  SELECT * FROM STREAM(samples.nyctaxi.trips);

CREATE OR REFRESH TEMPORARY STREAMING TABLE trips_data_quarantine(
  -- Option 1 - merge all expectations to have a single name in the pipeline event log
  CONSTRAINT quarantined_row EXPECT (pickup_zip IS NOT NULL OR dropoff_zip IS NOT NULL),
  -- Option 2 - Keep the expectations separate, resulting in multiple entries under different names
  CONSTRAINT invalid_pickup_zip EXPECT (pickup_zip IS NOT NULL),
  CONSTRAINT invalid_dropoff_zip EXPECT (dropoff_zip IS NOT NULL)
)
PARTITIONED BY (is_quarantined)
AS
  SELECT
    *,
    NOT ((pickup_zip IS NOT NULL) and (dropoff_zip IS NOT NULL)) as is_quarantined
  FROM STREAM(LIVE.raw_trips_data);

CREATE TEMPORARY LIVE VIEW valid_trips_data AS
SELECT * FROM LIVE.trips_data_quarantine WHERE is_quarantined=FALSE;

CREATE TEMPORARY LIVE VIEW invalid_trips_data AS
SELECT * FROM LIVE.trips_data_quarantine WHERE is_quarantined=TRUE;