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

以下は、利用可能なすべてのデータが満たすべき条件をカテゴリ別にまとめた一覧です。

演算子

Copied!
1
2
3
4
5
6
7
from transforms import expectations as E

E.true()           # 常に成功します
E.false()          # 常に失敗します
E.all(e1,e2,...)   # すべてのサブ条件を満たす場合に成功します
E.any(e1,e2,...)   # いずれかのサブ条件を満たす場合に成功します
E.negate(e1)       # サブ条件を満たさない場合に成功します

例：

Copied!
1
2
3
4
5
6
7
8
9
10
from transforms import expectations as E

E.all(
    E.col('a').gt(0),
    E.col('a').lt(100),
    E.any(
        E.col('b').gt(100),
        E.col('b').lt(0)
    )
)

列のデータが満たすべき条件

列のデータが満たすべき条件は E.col('column_name') で始まります。

any 演算子を使用すると、いずれかの列のデータが満たすべき条件を満たすかどうか、各行が個別にチェックされます。 たとえば、列 c1 の値が10より大きいか、0より小さいことを検証するには、次のようにします。

Copied!
1
2
3
4
5
6
from transforms import expectations as E

E.any(
    E.col("c1").lt(0),
    E.col("c1").gt(10)
)

より大きい、またはより小さい

Copied!
1
2
3
4
5
6
from transforms import expectations as E

E.col('c').gt(number|string)
E.col('c').gte(number|string)
E.col('c').lt(number|string)
E.col('c').lte(number|string)

例：

Copied!
1
2
3
from transforms import expectations as E

E.col(‘age').lt(120)

この条件は null 値を無視します（つまり、null 値は自動的に条件を満たすと判定されます）。null をチェックするには、E.col('col').non_null() を使用します。

列の比較

Copied!
1
2
3
4
5
6
7
8
from transforms import expectations as E

E.col('c1').equals_col('c2')  # 列 c1 の値が列 c2 の値に等しいです
E.col('c1').not_equals_col('c2')  # 列 c1 の値が列 c2 の値に等しくありません
E.col('c1').gt_col('c2')  # 列 c1 の値が列 c2 の値より大きいです
E.col('c1').gte_col('c2') # c1 の値が c2 の値以上です
E.col('c1').lt_col('c2')  # 列 c1 の値が列 c2 の値より小さいです
E.col('c1').lte_col('c2') # c1 の値が c2 の値以下です

プロパティの比較

Copied!
1
2
3
4
5
6
7
8
9
from transforms import expectations as E

E.col('c').null_percentage().lt(value)
E.col('c').null_count().gt(value)
E.col('c').distinct_count().equals(value)
E.col('c').approx_distinct_count().equals(value) # distinct_count のより高速なバージョンで、誤差の相対標準偏差が最大5%であることを保証します
E.col('c').sum().gt(value) # 数値列でのみ機能します
E.col('c').standard_deviation_sample().gt(value) # 数値列でのみ機能します
E.col('c').standard_deviation_population().gt(value) # 数値列でのみ機能します

例：

Copied!
1
2
3
4
5
6
7
8
9
from transforms import expectations as E

E.col("myCol").null_percentage().lt(0.1)    # myCol の null 値の割合が10%未満です
E.col("myCol").null_count().gt(100)         # myCol の null 値が100件より多いです
E.col("myCol").distinct_count().equals(5)   # myCol に5個の異なる値があります（バージョン 0.11.0 以降）
E.col('myCol').approx_distinct_count().equals(5).   # myCol に約5個の異なる値があります
E.col('myCol').sum().equals(5)              # myCol の値の合計が5です
E.col('myCol').standard_deviation_sample().gt(5) # myCol の標本標準偏差が5より大きいです
E.col('myCol').standard_deviation_population().gt(5) # myCol の母集団標準偏差が5より大きいです

等しい

Copied!
1
2
3
from transforms import expectations as E

E.col('c').equals(value)    # 列の値が入力に等しいです

例：

Copied!
1
2
3
from transforms import expectations as E

E.col('test_column').equals("success")

この条件は null 値を無視します（つまり、null 値は自動的に条件を満たすと判定されます）。null をチェックするには、E.col('col').non_null() を使用します。

等しくない

Copied!
1
2
3
from transforms import expectations as E

E.col('c').not_equals(value)    # 列の値が入力に等しくありません

例：

Copied!
1
2
3
from transforms import expectations as E

E.col('test_column').not_equals("failure")

この条件は null 値を無視します（つまり、null 値は自動的に条件を満たすと判定されます）。null をチェックするには、E.col('col').non_null() を使用します。

null

Copied!
1
2
3
4
from transforms import expectations as E

E.col('c').non_null()    # 列の値が null ではありません
E.col('c').is_null()     # 列の値が null です

指定値に含まれる

この条件は、列の値が許可された値のリストに含まれていることを検証します。 Array 列については、指定値に含まれる（配列）を参照してください。

Copied!
1
2
3
from transforms import expectations as E

E.col('c').is_in(a, b, ...)    # 列の値が指定されたリストに含まれます

rlike（正規表現）

pyspark.sql.functions.rlike と同様の、正規表現による部分一致です。

Copied!
1
2
3
from transforms import expectations as E

E.col('c').rlike(regex expression)    # 列の値が正規表現に一致します（部分一致）

例：

Copied!
1
2
3
from transforms import expectations as E

E.col('flight_number').rlike(r"^\D{2}\d{2,4}$")

この条件は null 値を無視します（つまり、null 値は自動的に条件を満たすと判定されます）。null をチェックするには、E.col('col').non_null() を使用します。

指定のデータ型である

Copied!
1
2
3
from transforms import expectations as E

E.col('c').has_type(Type)   # 列 'c' のデータ型が Type です

データ型に関する条件では、エンジンにかかわらず、シングルノードコンピューティングには Polars のデータ型 ↗を、PySpark コンピューティングには Spark SQL のデータ型 ↗を使用します。次の例のように、対応するライブラリをインポートし、データ型を適切に参照する必要があります。

例：

Copied!
1
2
3
4
from transforms import expectations as E
import polars as pl

E.col('age').has_type(pl.Int32)

Copied!
1
2
3
4
from transforms import expectations as E
from pyspark.sql import types as T

E.col('age').has_type(T.LongType())

存在する

Copied!
1
2
3
from transforms import expectations as E

E.col('c').exists()  # 列 'c' が出力データフレームに存在します

この条件は、指定した名前の列が出力データフレームに存在するかどうかをチェックします。 列のデータ型は任意です。

タイムスタンプのデータが満たすべき条件

タイムスタンプのデータが満たすべき条件は、データ型が Timestamp の列でのみ機能します。Date 列は現在サポートされていません。

固定タイムスタンプの比較

タイムスタンプ列の値を固定タイムスタンプと比較します。固定タイムスタンプは、ISO8601 形式の文字列または Python の datetime オブジェクトとして指定できます。 曖昧さを避けるため、すべてのタイムスタンプにタイムゾーン情報が必要です。

Copied!
1
2
3
4
5
6
from transforms import expectations as E

E.col("timestamp").is_after("2020-12-14T11:32:23+0000")
E.col("timestamp").is_before(datetime(2017, 11, 28, 23, 55, 59, 342380))
E.col("timestamp").is_on_or_after("2020-12-14T11:32:23+0000")
E.col("timestamp").is_on_or_before("2020-12-14T11:32:23+0000")

列間のタイムスタンプの比較

タイムスタンプ列の値を別のタイムスタンプ列の値と比較します。オフセット（整数の秒数）を任意で指定でき、指定すると、もう一方の列の値に加算されます。

比較は次のようになります。

first_column ($OPERATOR) second_column + offset_in_seconds

Copied!
1
2
3
4
5
6
7
8
9
from transforms import expectations as E

E.col("timestamp").is_after_col("second_timestamp")
E.col("timestamp").is_on_or_after_col("second_timestamp")
# 演算子は任意の offset_in_second 引数を受け取ります
# `second_timestamp` が `timestamp` より後で、その差が1時間未満であることを確認します
E.col("timestamp").is_before_col("second_timestamp", 3600)
# `second_timestamp` が `timestamp` より2時間を超えて前であることを確認します
E.col("timestamp").is_on_or_before_col("second_timestamp", -7200)

相対タイムスタンプの比較

タイムスタンプ列の値を、チェックの実行時刻（ビルドが行われる時刻など）にユーザー指定のオフセットを加えた時刻と比較します。オフセットは、整数の秒数または Python の timedelta オブジェクトとして指定できます。チェックがインスタンス化または実行される時刻自体が厳密ではないため、比較の精度は数分以内です。使用された正確なタイムスタンプは、チェック実行後にデータが満たすべき条件のチェック結果に表示されます。

主なメソッドとして、timestamp_offset_from_current_time と timestamp_offset_to_current_time の2つが用意されています。相対的な時間オフセットを自然に捉えやすくするため、2つの異なるメソッドを用意しています。そのため、引数としてサポートされるのは正の時間オフセットのみです。負のオフセットを使用する必要がある場合は、代わりにもう一方のメソッドの使用を検討してください。

timestamp_offset_from_current_time

このメソッドは、timestamp - now() が正の値になる、未来の相対時刻での使用を想定しています。この値が指定されたオフセットと比較されます。比較には、通常の比較演算子をすべて使用できます。

Copied!
1
2
3
4
5
6
7
8
from datetime import timedelta
from transforms import expectations as E

# タイムスタンプの値が現在より未来で、その差が1時間未満です
A = E.col("timestamp").timestamp_offset_from_current_time().lt(3600)

# タイムスタンプの値が現在より2時間を超えて未来です
B = E.col("timestamp").timestamp_offset_from_current_time().gt(timedelta(hours=2))

timestamp_offset_to_current_time

このメソッドは、now() - timestamp が正の値になる、過去の相対時刻での使用を想定しています。この値が指定されたオフセットと比較されます。比較には、通常の比較演算子をすべて使用できます。

Copied!
1
2
3
4
5
6
7
8
from datetime import timedelta
from transforms import expectations as E

# タイムスタンプの値が現在より過去で、その差が90分未満です
C = E.col("timestamp").timestamp_offset_to_current_time().lt(5400)

# タイムスタンプの値が現在より2時間を超えて過去です
D = E.col("timestamp").timestamp_offset_to_current_time().gt(timedelta(hours=2))

例：期待される結果

チェックが1月1日の午後4時に実行されると仮定すると、タイムスタンプの値ごとに、上記のチェックで期待される結果は次のとおりです。

   < ------- PAST ---------------------  NOW  -------------------- FUTURE ------>
   |    1pm   |    2pm   |    3pm   |    4pm   |    5pm   |    6pm   |    7pm   |
---+----------+----------+----------+----------+----------+----------+----------+
 A |   PASS   |   PASS   |   PASS   |   PASS   |   FAIL   |   FAIL   |   FAIL   |
---+----------+----------+----------+----------+----------+----------+----------+
 B |   FAIL   |   FAIL   |   FAIL   |   FAIL   |   FAIL   |   FAIL*  |   PASS   |
---+----------+----------+----------+----------+----------+----------+----------+
 C |   FAIL   |   FAIL   |   PASS   |   PASS   |   PASS   |   PASS   |   PASS   |
---+----------+----------+----------+----------+----------+----------+----------+
 D |   PASS   |   FAIL*  |   FAIL   |   FAIL   |   FAIL   |   FAIL   |   FAIL   |
---+----------+----------+----------+----------+----------+----------+----------+

チェック B と D の比較は、等しい値を含まない厳密な比較です。等しい値も含める比較には、ge と le を使用します。

グループとプロパティのタイムスタンプの比較

ほとんどのタイムスタンプ比較は、通常のデータフレームまたはグループ化されたデータフレームの派生プロパティでも使用できます。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from datetime import timedelta
from transforms import expectations as E

# 最大のタイムスタンプが指定された固定の日付より後であることを確認します
E.col("timestamp").max_value().is_after("2020-12-14T12:23:50+0000")

# 最も古いタイムスタンプが現在より過去で、その差が1日未満であることを確認します
E.col("timestamp").min_value().timestamp_offset_to_current_time().lt(timedelta(days=1))

# 各カテゴリーの最後の日付が現在より2か月を超えて未来であることを確認します
E.group_by("category")
    .col("timestamp")
    .max_value()
    .timestamp_offset_from_current_time()
    .gt(timedelta(months=2))

配列のデータが満たすべき条件

すべての条件が配列型の列で機能するわけではありません。配列型の列では、以下で説明する専用の条件のみを使用できます。

配列が指定値に含まれる

is_in 条件は、配列型の列でも機能します。 配列の場合、この条件は、配列に is_in 句で指定した値のみが含まれていることをテストします。

Copied!
1
2
3
from transforms import expectations as E

E.col('array_col').is_in('a', 'b', 'c') # 'array_col' 内のどの配列も、値 'a'、'b'、'c' のみを含むことができます。

配列が指定値を含む

array_contains 条件では、配列列の各行に特定の値が含まれていることをチェックできます。

Copied!
1
2
3
from transforms import expectations as E

E.col('array_col').array_contains('a') # すべての行の 'array_col' に値 'a' が含まれている必要があります。

配列サイズ

size 条件では、各行の配列が特定のサイズであることをチェックできます。

Copied!
1
2
3
4
from transforms import expectations as E

E.col('array_col').size().gt(1) # 'array_col' の長さが1より大きい必要があります。
E.col('array_col').size().equals(2) # 'array_col' の長さが2に等しい必要があります。

グループ化したデータが満たすべき条件

グループ化したデータが満たすべき条件は E.group_by('column_1', 'column_2', ...) で始まります。グループ化したデータが満たすべき条件では、列の組み合わせに対して条件を設定できます。

一意である

Copied!
1
2
3
from transforms import expectations as E

E.group_by('col1', 'col2').is_unique()   # 列の値を組み合わせたとき、その組み合わせがデータセット内で一意です

行数

行数の条件は、グループごとの行数をテストします。group_by が空の場合は、データセット全体の行数に対してテストします。

Copied!
1
2
3
4
5
from transforms import expectations as E

E.group_by('col1', 'col2').count().gt(100)    # 'col1'、'col2' でグループ化した各グループの行の件数が100より大きい必要があります
E.group_by().count().lt(100)    # データセットの行の件数が100未満です。
E.count().equals(0)    # 空の group_by の省略形です。データセットの行の件数が0に等しいことを検証します。

グループ化したプロパティが満たすべき条件

すべてのプロパティの比較条件は、グループ化したデータが満たすべき条件としても使用できます。

Copied!
1
2
3
4
from transforms import expectations as E

E.group_by('col1').col('value_col').distinct_count().equals(3) # 'col1' でグループ化した各グループの 'value_col' の異なる値の件数が3に等しい必要があります。
E.group_by('col1').col('value_col').null_percentage().lt(0.5) # 'col1' でグループ化した各グループの 'value_col' の null 値の割合が50%未満である必要があります。

主キー

主キーの条件は、1個以上の列名を受け取り、次のことを検証します。

1. 各列に null 値がないこと
2. 列の組み合わせが一意であること

Copied!
1
2
3
4
from transforms import expectations as E

E.primary_key('c1')             # 列 `c1` が一意で、null ではありません。
E.primary_key('c1', 'c2',...)   # 列 {'c1', 'c2',...} はそれぞれ null ではなく、組み合わせが一意です

例：

Copied!
1
2
3
from transforms import expectations as E

E.primary_key('time', 'event')

スキーマが満たすべき条件

すべてのスキーマが満たすべき条件は E.schema() で始まります。

トランスフォームで使用するコンピューティングエンジンに基づいて、適切なデータ型のチェックを使用します。デフォルトのシングルノード（軽量）トランスフォームでは、Polars のデータ型 ↗によるチェックを使用します。分散（Spark）トランスフォームでは、PySpark SQL のデータ型 ↗によるチェックを使用します。

Copied!
1
2
3
4
5
6
7
from transforms import expectations as E

E.schema().contains({'col_name':type})  # データセットの列に、列挙された列が含まれている必要があります。
E.schema().equals({'col_name':type})   # データセットの列が列挙された列に完全一致します（追加の列はありません）
E.schema().is_subset_of({'col_name':type})   # データセットの列が、列挙された列の部分集合である必要があります。
                                             # データセット内のすべての列がチェックで定義されている必要があります。
                                             # データセットに存在しない列もチェックで定義できます。

例：

Copied!
1
2
3
4
5
6
7
8
9
from transforms import expectations as E
import polars as pl

E.schema().contains(
    {
        'name': pl.String(),
        'int_list': pl.List(pl.Int32)
    }
)

Copied!
1
2
3
4
5
6
7
8
9
from transforms import expectations as E
from pyspark.sql import types as T

E.schema().contains(
    {
        'id': T.IntegerType(),
        'name': T.StringType()
    }
)

条件付き

条件付きのデータが満たすべき条件は、3つの条件を受け取り、次のことを検証します。

1. when 条件を満たす行が、then 条件も満たすこと
2. when 条件を満たさない行が、otherwise 条件を満たすこと

Copied!
1
2
3
4
5
6
7
8
from transforms import expectations as E

E.when(
    when_exp,
    then_exp
).otherwise(
    otherwise_exp
)

たとえば、「myCol」が0より大きい場合、「myOtherCol」は ["a"] に含まれている必要があり、それ以外の場合、「myOtherCol」は ["b"] に含まれている必要があります。

Copied!
1
2
3
4
5
6
7
8
from transforms import expectations as E

E.when(
    E.col("myCol").gt(0),
    E.col("myOtherCol").is_in("a")
).otherwise(
    E.col("myOtherCol").is_in("b")
)

E.true() と E.false() を使用して、条件付きのデータが満たすべき条件の otherwise 分岐に単純なデフォルトを設定します。

外部値に関するデータが満たすべき条件［試験的］

外部値に関するデータが満たすべき条件は、異なるデータセット間のデータの関係を検証します。これらの条件には結合が伴い、評価コストが非常に高くなる場合があるため、慎重に使用してください。

E.dataset_ref() はトランスフォームの入力名を受け取ります。これは、トランスフォームのデコレーター内の =Input(...) の前にあるキーワードで、対応する計算関数のパラメーターとしても使用されます。データセットの RID、Compass パス、表示名は受け取りません。参照するデータセットは、トランスフォームの入力である必要があります。

参照整合性

このデータが満たすべき条件は、検証対象のデータセットで選択した列のすべての値が、外部データセットの指定した列に存在することを検証します。null 値は無視されます。

E.dataset_ref('other_dataset_name').col('f_col') を使用して、外部列を他方のデータセットの入力名で修飾します。

列参照の使用方法は、is_in() の使用方法と似ています。

Copied!
1
E.col('pk').is_in_foreign_col(E.dataset_ref('other_dataset').col('fk'))

次の例では、E.dataset_ref("countries") が countries=Input(...) 宣言および計算関数の countries パラメーターに対応しています。

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


@transform.using(
    output=Output(
        "/path/to/output",
        checks=Check(
            E.col("country_code").is_in_foreign_col(
                E.dataset_ref("countries").col("country_code")
            ),
            "Country code exists",
            on_error="FAIL",
        ),
    ),
    transactions=Input("/path/to/transactions"),
    countries=Input("/path/to/countries"),
)
def compute(output, transactions, countries):
    output.write_table(transactions.polars())

データセット間の行数比較

データセット間の行数比較を使用すると、あるデータセットの行数を別のデータセットの行数と比較できます。

たとえば、出力の行数が入力データセットの行数と同じであることを確認できます。

Copied!
1
E.count().equals(E.dataset_ref('input_dataset_name').count())

ここで、input_dataset_name は上記で説明したトランスフォームの入力名です。完全な例については、参照整合性を参照してください。

データセットの行数比較には、次の演算子を使用できます。

Copied!
1
2
3
4
5
6
7
from transforms import expectations as E

E.count().equals(E.dataset_ref('input_dataset_name').count()) # 等しい
E.count().lt(E.dataset_ref('input_dataset_name').count()) # より小さい
E.count().lte(E.dataset_ref('input_dataset_name').count()) # 以下
E.count().gte(E.dataset_ref('input_dataset_name').count()) # 以上
E.count().gt(E.dataset_ref('input_dataset_name').count()) # より大きい