注: 以下の翻訳の正確性は検証されていません。AIPを利用して英語版の原文から機械的に翻訳されたものです。

はじめに

このガイドでは、Python トランスフォームのリポジトリでデータが満たすべき条件を設定する方法を説明します。データが満たすべき条件の概要については、ドキュメントを参照してください。

リポジトリのセットアップ: transforms-expectations ライブラリ

コードリポジトリの左側にあるライブラリ検索パネルを開きます。transforms-expectations ライブラリはすでにインストールされているはずです。インストール済みライブラリの一覧を確認して、これを検証します。transforms-expectations がまだインストールされていない場合は、ここで検索してインストールします。

トランスフォームの設定

トランスフォームファイルに expectations と Check をインポートします。

Copied!1
2
from transforms.api import transform_df, Input, Output, Check
from transforms import expectations as E

一般的なスキーマや列に関するデータが満たすべき条件には、適切なデータ型ライブラリをインポートします。

データ型に関するデータが満たすべき条件では、単一ノードコンピューティングの場合はエンジンにかかわらず Polars のデータ型 ↗を使用し、PySpark コンピューティングの場合は Spark SQL のデータ型 ↗を使用します。次の例に示すように、該当するライブラリをインポートし、データ型を適切に参照する必要があります。

Copied!1
2
3
import polars as pl

# 例: E.col('age').has_type(pl.Int32)

Copied!1
2
3
from pyspark.sql import types as T

# 例: E.col('age').has_type(T.LongType())

チェックを作成する

単一のチェックの基本構造：

Copied!1
Check(expectation, 'Check unique name', on_error='WARN/FAIL')

データが満たすべき条件 - 単一のデータが満たすべき条件です。複数の下位条件を組み合わせた複合条件（たとえば any/all 演算子を使用）にもできます。
チェックの一意の名前 - トランスフォーム内で一意である必要があります（出力と入力で同じ名前を共有することはできません）。アプリ間（たとえば Data Health、Builds アプリケーション）でチェックを識別するために使用されます。
on_error - データが満たすべき条件が満たされない場合のジョブの動作を定義します。
- FAIL（デフォルト）- チェックに失敗するとジョブが中止されます。
- WARN - ジョブは続行され、警告が生成されて Data Health によって処理されます。

単一ノードトランスフォームの仮想テーブル出力では、on_error='FAIL' の設定に対応していません。

データセットにチェックを割り当てる

各チェックは、単一の入力または出力に渡す必要があります。単一のチェックは checks=check1 として渡し、複数のチェックは配列で渡します：checks=[check1, check2, ...]

複数のチェックを使用すると、データが満たすべき条件の構造が読みやすくなり、意味のあるチェックごとに動作を個別に制御できます。

出力に対する単純な主キーチェックの例：

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E
import polars as pl


@transform.using(
    output=Output(
        "/path/dataset",
        checks=Check(E.primary_key('id'), 'Primary Key', on_error='FAIL')
    ),
    input=Input("/path/input"),
)
def compute(output, input):
    input_df = input.polars()
    return output.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E


@transform.using(
    output=Output(
        "/path/dataset",
        checks=Check(E.primary_key('id'), 'Primary Key', on_error='FAIL')
    ),
    input=Input("/path/input"),
)
def compute(ctx, output, input):
    query = ctx.duckdb().conn.sql("SELECT * FROM input")
    return output.write_table(query)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E
import pandas as pd


@transform.using(
    output=Output(
        "/path/dataset",
        checks=Check(E.primary_key('id'), 'Primary Key', on_error='FAIL')
    ),
    input=Input("/path/input"),
)
def compute(output, input):
    input_df = input.pandas()
    return output.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
from transforms.api import transform_df, Input, Output, Check
from transforms import expectations as E


@transform_df(
    Output(
        "/path/dataset",
        checks=Check(E.primary_key('id'), 'Primary Key', on_error='FAIL')
    ),
    input=Input("/path/input")
)
def my_compute_function(input):
    return input

複雑なチェック

複合条件を使用して、より複雑なチェックを追加することもできます。たとえば、列 age が null ではなく、指定した範囲内にあることをチェックしてみましょう。複合条件を定義してトランスフォーム内の複数のチェックで使用し、エラー時の動作をそれぞれ変えられる点に注目してください。

チェックは、複合条件で構成されている場合でも、全体として監視されます。複合条件の特定の部分を監視する（つまり、状態を確認して通知を受け取る）場合は、複数の異なるチェックに分割することを推奨します。

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E
import polars as pl


# 年齢が0から200の間であれば、有効とみなします。
expect_valid_age = E.all(
    E.col('age').non_null(),
    E.col('age').gte(0),
    E.col('age').lt(200)
)


@transform.using(
    output=Output(
        "/path/dataset",
        checks=[
            Check(E.primary_key('id'), 'Primary Key', on_error='FAIL'),
            Check(expect_valid_age, 'Valid age on output', on_error='FAIL')
        ]
    ),
    input=Input(
        "Users/data/input",
        checks=Check(expect_valid_age, 'Valid age on input', on_error='WARN')
    )
)
def compute(output, input):
    input_df = input.polars()
    return output.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E


# 年齢が0から200の間であれば、有効とみなします。
expect_valid_age = E.all(
    E.col('age').non_null(),
    E.col('age').gte(0),
    E.col('age').lt(200)
)


@transform.using(
    output=Output(
        "/path/dataset",
        checks=[
            Check(E.primary_key('id'), 'Primary Key', on_error='FAIL'),
            Check(expect_valid_age, 'Valid age on output', on_error='FAIL')
        ]
    ),
    input=Input(
        "Users/data/input",
        checks=Check(expect_valid_age, 'Valid age on input', on_error='WARN')
    )
)
def compute(ctx, output, input):
    conn = ctx.duckdb().conn
    query = conn.sql("SELECT * FROM input")
    return output.write_table(query)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from transforms.api import transform, Input, Output, Check
from transforms import expectations as E
import pandas as pd


# 年齢が0から200の間であれば、有効とみなします。
expect_valid_age = E.all(
    E.col('age').non_null(),
    E.col('age').gte(0),
    E.col('age').lt(200)
)


@transform.using(
    output=Output(
        "/path/dataset",
        checks=[
            Check(E.primary_key('id'), 'Primary Key', on_error='FAIL'),
            Check(expect_valid_age, 'Valid age on output', on_error='FAIL')
        ]
    ),
    input=Input(
        "Users/data/input",
        checks=Check(expect_valid_age, 'Valid age on input', on_error='WARN')
    )
)
def compute(output, input):
    input_df = input.pandas()
    return output.write_table(input_df)

Copied!1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from transforms.api import transform_df, Input, Output, Check
from transforms import expectations as E


# 年齢が0から200の間であれば、有効とみなします。
expect_valid_age = E.all(
    E.col('age').non_null(),
    E.col('age').gte(0),
    E.col('age').lt(200)
)


@transform_df(
    Output(
        "/Users/data/dataset",
        checks=[
            Check(E.primary_key('id'), 'Primary Key', on_error='FAIL'),
            Check(expect_valid_age, 'Valid age on output', on_error='FAIL')
        ]
    ),
    input=Input(
        "Users/data/input",
        checks=Check(expect_valid_age, 'Valid age on input', on_error='WARN')
    )
)
def my_compute_function(input):
    return input

←

PREVIOUS仮想テーブルとコンピュートプッシュダウン / Snowflake へのコンピュートプッシュダウン

NEXTデータが満たすべき条件のリファレンス

→