リファレンス

以下は、利用可能なすべてのデータ期待値のカテゴリー別リストです。

演算子

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
11
12
13
14
15
# transforms ライブラリから expectations を E としてインポートします
from transforms import expectations as E

# 以下のすべての期待値を設定します：
E.all(
    # 'a'という名前の列の値が0より大きい
    E.col('a').gt(0),
    # 'a'という名前の列の値が100より小さい
    E.col('a').lt(100),
    # 'b'という名前の列の値が100より大きいか、または0より小さい
    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
7
from transforms import expectations as E

# "c1"カラムの値が0より小さいか、または10より大きいかどうかをチェックする
E.any(
    E.col("c1").lt(0),  # "c1"カラムの値が0より小さいかチェック
    E.col("c1").gt(10)  # "c1"カラムの値が10より大きいかチェック
)

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

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

E.col('c').gt(number|string)  # 'c'列の値が(数値|文字列)より大きい
E.col('c').gte(number|string) # 'c'列の値が(数値|文字列)以上
E.col('c').lt(number|string)  # 'c'列の値が(数値|文字列)より小さい
E.col('c').lte(number|string) # 'c'列の値が(数値|文字列)以下

例えば:

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

# E.col('age') は、'age' カラムに対する操作を意味します。
# lt(120) は、'age' カラムの値が 120 より小さいことをチェックします。
E.col('age').lt(120)

行の比較

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
10
11
12
13
14
15
16
from transforms import expectations as E

# c列のNULLの割合がvalueより小さい
E.col('c').null_percentage().lt(value)
# c列のNULLの個数がvalueより大きい
E.col('c').null_count().gt(value)
# c列の異なる値の数がvalueと等しい
E.col('c').distinct_count().equals(value)
# エラーの相対標準偏差が最大5%の高速バージョンのc列の異なる値の数がvalueと等しい
E.col('c').approx_distinct_count().equals(value)
# 数値列のみで動作し、c列の合計がvalueより大きい
E.col('c').sum().gt(value)
# 数値列のみで動作し、c列の標本標準偏差がvalueより大きい
E.col('c').standard_deviation_sample().gt(value)
# 数値列のみで動作し、c列の母集団標準偏差が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
4
from transforms import expectations as E

# 'test_column'という名前のカラムが"success"と等しいかどうかをチェック
E.col('test_column').equals("success")

等しくない

Copied!
1
2
3
from transforms import expectations as E

E.col('c').not_equals(value)    # カラムの値が入力と等しくない

たとえば:

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

# 'test_column'列が"failure"と等しくないことを検証します
E.col('test_column').not_equals("failure")

Null

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

E.col('c').non_null()    # カラムの値がnullではない
E.col('c').is_null()     # カラムの値がnullである

一覧内にある

この期待値は、行の値が承認された値のリスト内にあることを確認します。 配列行については、is in (array) を参照してください。

Copied!
1
2
3
from transforms import expectations as E

E.col('c').is_in(a, b, ...)    # カラム値が与えられたリストに含まれている

rlike (regex)

Regex の部分一致で、pyspark.sql.functions.rlike に似ています。

Copied!
1
2
3
from transforms import expectations as E

E.col('c').rlike(regex expression)    # カラムの値が正規表現と一致する（部分一致）

例えば:

Copied!
1
2
3
4
5
6
# transforms モジュールから expectations を E としてインポートします
from transforms import expectations as E

# 'flight_number'という列に対して正規表現を使用します
# 正規表現 "^\D{2}\d{2,4}$" は、先頭が2文字の非数字（\D{2}）で、その後に2~4桁の数字（\d{2,4}）が続く、というパターンを表しています
E.col('flight_number').rlike(r"^\D{2}\d{2,4}$")

型がある

Copied!
1
2
3
from transforms import expectations as E

E.col('c').has_type(Type)   # 列 'c' の型は Type です

この期待は、pyspark.sql から types をインポートする必要があります（例：as T）

例：

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

# 'age'というカラムがLongType（整数型）であることを期待する
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
7
8
9
10
11
12
13
from transforms import expectations as E

# "timestamp"カラムの日付が"2020-12-14T11:32:23+0000"より後であることを確認します
E.col("timestamp").is_after("2020-12-14T11:32:23+0000")

# "timestamp"カラムの日付が"2017-11-28 23:55:59.342380"より前であることを確認します
E.col("timestamp").is_before(datetime(2017, 11, 28, 23, 55, 59, 342380))

# "timestamp"カラムの日付が"2020-12-14T11:32:23+0000"以降であることを確認します
E.col("timestamp").is_on_or_after("2020-12-14T11:32:23+0000")

# "timestamp"カラムの日付が"2020-12-14T11:32:23+0000"以前であることを確認します
E.col("timestamp").is_on_or_before("2020-12-14T11:32:23+0000")

タイムスタンプ行の比較

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

比較は以下のようになります:

first_column（$OPERATOR）second_column + オフセット秒数
# first_column（最初の列）と second_column（次の列）を $OPERATOR で演算して、オフセット秒数を足す

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

# "timestamp"が"second_timestamp"より後であることを確認します
E.col("timestamp").is_after_col("second_timestamp")
# "timestamp"が"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)

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

タイムスタンプの行の値を、チェックが実行される時間（例えばビルドが行われる時間）とユーザーが指定したオフセットに対して比較します。オフセットは秒数で表された整数、または timedelta Pythonオブジェクトとして提供できます。

主に提供される2つのメソッドは、timestamp_offset_from_current_timeとtimestamp_offset_to_current_timeです。私たちは、相対時間オフセットに対して自然な方法で推論するのを助けるために、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)  # タイムスタンプの値が現在時間から5400秒（90分）以内であることを確認

# タイムスタンプの値が2時間以上前
D = E.col("timestamp").timestamp_offset_to_current_time().gt(timedelta(hours=2))  # タイムスタンプの値が現在時間から2時間以上経過していることを確認

例：期待される結果

チェックが1月1日の午後4時に実行されると仮定すると、timestampの異なる値に対して、上記のチェックの期待される結果は以下の通りです。

   < ------- 過去 ---------------------  現在  -------------------- 未来 ------>
   |    1pm   |    2pm   |    3pm   |    4pm   |    5pm   |    6pm   |    7pm   |
---+----------+----------+----------+----------+----------+----------+----------+
 A |   合格   |   合格   |   合格   |   合格   |   失敗   |   失敗   |   失敗   |
---+----------+----------+----------+----------+----------+----------+----------+
 B |   失敗   |   失敗   |   失敗   |   失敗   |   失敗   |   失敗*  |   合格   |
---+----------+----------+----------+----------+----------+----------+----------+
 C |   失敗   |   失敗   |   合格   |   合格   |   合格   |   合格   |   合格   |
---+----------+----------+----------+----------+----------+----------+----------+
 D |   合格   |   失敗*  |   失敗   |   失敗   |   失敗   |   失敗   |   失敗   |
---+----------+----------+----------+----------+----------+----------+----------+

訳注：

• 各行（A, B, C, D）は異なるテストケースを示しています。
• 各列（1pm, 2pm, 3pm, ...）は時間を示しています。過去から未来にかけてのテスト結果を時間ごとに追跡しています。
• "PASS"はテストが成功したことを示しています。
• "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
4
from transforms import expectations as E

# 'array_col'の配列は、値が'a'、'b'、'c'のいずれかでなければならない。
E.col('array_col').is_in('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('行_1', '行_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. 各行に 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
4
from transforms import expectations as E

# 'time'と'event'をプライマリーキーとして設定
E.primary_key('time', 'event')

スキーマの期待値

すべてのスキーマ期待値は E.schema() で始まります。

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
10
# transformsというライブラリからexpectationsというモジュールをインポートします
from transforms import expectations as E

# E.schemaメソッドを使用してスキーマを定義します。ここでは'id'が整数型、'name'が文字列型となるように設定しています。
E.schema().contains(
    {
        'id': T.IntegerType(),
        'name': T.StringType()
    }
)

この期待値は、pyspark.sql から types をインポートする必要があります（例：as T）

条件付き

条件付きの期待値は、3つの期待値を取り、以下を確認します：

1. when-expectationを通過する行もthen-expectationを通過します
2. when-expectationに失敗する行は、otherwise-expectationsを通過します

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

# `when_exp`がTrueの場合、`then_exp`を実行
# それ以外の場合、`otherwise_exp`を実行
E.when(
    when_exp,
    then_exp
).otherwise(
    otherwise_exp
)

例えば、"myCol"が0より大きい場合、"myOtherCol"は["a"]になければならず、それ以外の場合は"myOtherCol"は["b"]になければなりません。

Copied!
1
2
3
4
5
6
7
8
9
10
11
# 日本語のコメントを追加します
from transforms import expectations as E

# myColが0より大きい場合、myOtherColが"a"に含まれることを期待
# そうでない場合、myOtherColが"b"に含まれることを期待
E.when(
    E.col("myCol").gt(0),
    E.col("myOtherCol").is_in("a")
).otherwise(
    E.col("myOtherCol").is_in("b")
)

外部値の期待値

警告：これは実験的な機能です。

外部値の期待値は、異なるデータセット間のデータの関係を検証します。これらの期待値には結合が含まれており、評価するのに非常にコストがかかる可能性があるため、慎重に使用してください。

参照整合性

この期待値は、期待されるデータセットの選択された行のすべての値が外部データセットの指定された行に存在することを検証します。Null 値は無視されます。

一致させるための外部行は、他のデータセットの名前を使用して作成されたデータセット参照によって認定されます：E.dataset_ref('other_dataset_name').col('f_col')。

外部データセットは必ずユーザーの変換の入力でなければならず、E.dataset_ref の参照はそれが割り当てられている変数の名前でなければなりません。

行の参照の使用は is_in() の使用に似ています：

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

主キー（'pk'）が外部キー（'fk'）の列内に存在するかどうかを確認

クロスデータセットの行数比較

クロスデータセットの行数比較は、あるデータセットの行数と別のデータセットの行数を比較するために使用できます。

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

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

出力データセットの行数が入力データセットの行数と同じであることを確認する

比較するデータセットは、他のデータセットの名前を使用して作成したデータセット参照で限定されます。

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

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()) # より大きい