Delta Live Tablesのデータ品質制約

Delta Live Tables data quality constraints | Databricks on AWS [2022/1/10]の翻訳です。

本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。

データセットのコンテンツに対するデータ品質制約を定義するためにエクスペクテーションを用いることができます。エクスペクテーションは説明文、条件、そして条件を満たさない場合に行うアクションから構成されます。Pythonのデコレーター、SQLのconstraint句を用いて、クエリーに対するエクスペクテーションを適用することができます。

単一のデータ品質制約を定義するには、PythonあるいはSQLクエリーでexpect、expect or drop、expect or failエクスペクテーションを使用します。

@expect_all、@expect_all_or_drop、@expect_all_or_failデコレーターを用いることで、Pythonパイプラインにおいて1つ以上のデータ品質制約を含むエクスペクテーションを定義することができます。これらのデコレーターは引数としてキーがエクスペクテーション名、値がエクスペクテーションの制約となるPythonのディクショナリーを受け取ります。

Delta Live Tablesのイベントログをクエリーすることで、エクスペクテーションに違反したレコードの数のようなデータ品質のメトリクスを参照することができます。

不正なレコードを保持する

エクスペクテーションに違反したレコードを保持したい場合には、expectオペレーターを使用します。エクスペクテーションに違反したレコードは、適切なレコードと共にターゲットデータセットに追加されます。

Python

@dlt.expect("valid timestamp", "col(“timestamp”) > '2012-01-01'")

SQL

CONSTRAINT valid_timestamp EXPECT (timestamp > '2012-01-01')

不正なレコードを削除する

不正なレコードの処理を行わないようにするには、expect or dropオペレーターを使用します。エクスペクテーションに違反したレコードはターゲットのデータセットから削除されます。

Python

@dlt.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")

SQL

CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL and current_page_title IS NOT NULL) ON VIOLATION DROP ROW

不正なレコードで処理を失敗させる

不正なレコードを受け付けられない場合は、レコードが検証を通過しなかった際に即座に実行を停止するために、expect or failを使用します。オペレーションがテーブルのアップデートの場合、システムはトランザクションを原子的にロールバックします。

Python

@dlt.expect_or_fail("valid_count", "count > 0")

SQL

CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

エクスペクテーションに違反したためにパイプラインを失敗させた場合、パイプラインを再実行する前に不正なデータを適切に取り扱うように、パイプラインのコードを修正しなくてはなりません。

Failのエクスペクテーションは、違反を検知し、レポートするために必要な情報を追跡するために、変換処理のSparkクエリープランを修正します。多くのクエリーに対して、違反を引き起こした入力レコードを識別するためにこの情報を活用することができます。以下の例がサンプルの例外となります。

Expectation Violated:
{
  "flowName": "a-b",
  "verboseInfo": {
    "expectationsViolated": [
      "x1 is negative"
    ],
    "inputData": {
      "a": {"x1": 1,"y1": "a },
      "b": {
        "x2": 1,
        "y2": "aa"
      }
    },
    "outputRecord": {
      "x1": 1,
      "y1": "a",
      "x2": 1,
      "y2": "aa"
    },
    "missingInputData": false
  }
}

複数のエクスペクテーション

検証に失敗したレコードをターゲットデータセットに含める際に、複数のデータ品質制約を指定するには、expect_allを使用します。

Python

@dlt.expect_all({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

検証に失敗したレコードをターゲットデータセットから削除する際に、複数のデータ品質制約を指定するには、expect_all_or_dropを使用します。

Python

@dlt.expect_all_or_drop({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

レコードが検証に失敗したレ際にパイプラインの実行を停止する場合に複数のデータ品質制約を指定するには、expect_all_or_failを使用します。

Python

@dlt.expect_all_or_fail({"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"})

また、変数としてエクスペクテーションのコレクションを定義し、パイプラインに1つ以上のクエリーを指定することができます。

Python

valid_pages = {"valid_count": "count > 0", "valid_current_page": "current_page_id IS NOT NULL AND current_page_title IS NOT NULL"}

@dlt.table
@dlt.expect_all(valid_pages)
def raw_data():
  # Create raw dataset

@dlt.table
@dlt.expect_all_or_drop(valid_pages)
def prepared_data():
  # Create cleaned and prepared dataset

Databricks 無料トライアル

Databricks 無料トライアル