Snowparkチェックポイントライブラリ: Hypothesis¶

仮説単体テスト¶

仮説 は Python 用の強力なテストライブラリで、さまざまな入力データを自動的に生成することで、従来のユニットテストを強化するように設計されています。Hypothesisはプロパティベースのテストを使用し、個々のテストケースを指定する代わりに、コードの期待される動作をプロパティまたは条件で記述し、Hypothesisがそれらのプロパティを徹底的にテストする例を生成します。このアプローチは、エッジケースや予期せぬ動作を発見するのに役立ち、複雑な関数に特に効果的です。詳しくは 仮説 を参照してください。

snowpark-checkpoints-hypothesis パッケージは Hypothesis ライブラリを拡張し、テスト用に合成 Snowpark DataFrames を生成します。多様でランダムなテストデータを生成するHypothesisの機能を活用することで、さまざまなスキーマと値を持つSnowpark DataFrames を作成し、実世界のシナリオをシミュレートしてエッジケースを発見し、ロバストなコードを保証して複雑な変換の正しさを検証することができます。

Snowparkの仮説戦略は、合成データの生成にPanderaを利用しています。 dataframe_strategy 関数は、指定されたスキーマを使用して、それに準拠した Pandas DataFrame を生成し、それを Snowpark DataFrame に変換します。

関数の署名:

def dataframe_strategy(
  schema: Union[str, DataFrameSchema],
  session: Session,
  size: Optional[int] = None
) -> SearchStrategy[DataFrame]

関数のパラメーター：

• schema: 生成されるSnowparkデータフレームが一致すべき列、データタイプ、チェックを定義するスキーマ。スキーマは次のようになります：

  • snowpark-checkpoints-collectors パッケージの collect_dataframe_checkpoint 関数によって生成された JSON スキーマファイルへのパス。

  • pandera.api.pandas.containerDataFrameSchema のインスタンス。

• session: snowflake.snowpark.Session のインスタンスで、Snowpark DataFrames の作成に使用されます。

• size: 各Snowpark の生成行数 DataFrame。このパラメーターが提供されない場合、ストラテジーは異なるサイズの DataFrames を生成します。

関数出力：

Snowpark DataFrames を生成する Hypothesis SearchStrategy を返します。

サポートされているデータ型とサポートされていないデータ型¶

dataframe_strategy 関数は、Snowpark DataFrames のさまざまなデータ タイプの生成をサポートします。関数に渡されるスキーマ引数のタイプによって、ストラテジーがサポートするデータ型は異なります。ストラテジーがサポートされていないデータ型を見つけた場合は例外が発生することに注意してください。

次の表は、 JSON ファイルを schema 引数として渡した場合に、 dataframe_strategy 関数でサポートされる PySpark データファイルとサポートされない データファイルのタイプを示しています。

schema の引数として DataFrameSchema オブジェクトを渡した場合に dataframe_strategy 関数でサポートされる Pandera データ型と、それらがマッピングされる Snowpark データ型を次の表に示します。

例¶

Hypothesisライブラリを使用してSnowpark DataFrames を生成する典型的なワークフローは以下のとおりです。

1. Python の標準的なテスト関数を作成し、あなたのコードがすべての入力に対して満たすべきさまざまなアサーションや条件を指定します。

2. Hypothesis @given デコレーターをテスト関数に追加し、 dataframe_strategy 関数を引数として渡します。 @given デコレーターの情報については、 仮説.given を参照してください。

3. テスト関数を実行します。テストが実行されると、Hypothesisは生成された入力をテストの引数として自動的に提供します。

例1: JSON ファイルから Snowpark DataFrames を生成します。

以下は、 snowpark-checkpoints-collectors パッケージの collect_dataframe_checkpoint 関数によって生成された JSON スキーマファイルから Snowpark DataFrames を生成する方法の例です。

from hypothesis import given

from snowflake.hypothesis_snowpark import dataframe_strategy
from snowflake.snowpark import DataFrame, Session


@given(
    df=dataframe_strategy(
        schema="path/to/file.json",
        session=Session.builder.getOrCreate(),
        size=10,
    )
)
def test_my_function_from_json_file(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...

例 2: Pandera DataFrameSchema オブジェクトから Snowpark DataFrame を生成します。

以下は、Pandera DataFrameSchema のインスタンスから Snowpark DataFrames を生成する方法の例です。詳しくは Pandera DataFrameSchema を参照してください。

import pandera as pa

from hypothesis import given

from snowflake.hypothesis_snowpark import dataframe_strategy
from snowflake.snowpark import DataFrame, Session


@given(
    df=dataframe_strategy(
        schema=pa.DataFrameSchema(
            {
                "boolean_column": pa.Column(bool),
                "integer_column": pa.Column("int64", pa.Check.in_range(0, 9)),
                "float_column": pa.Column(pa.Float32, pa.Check.in_range(10.5, 20.5)),
            }
        ),
        session=Session.builder.getOrCreate(),
        size=10,
    )
)
def test_my_function_from_dataframeschema_object(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...

例3： 仮説動作のカスタマイズ

仮説 @settings デコレーターを使って、テストの動作をカスタマイズすることもできます。このデコレーターを使うと、さまざまな構成パラメーターをカスタマイズして、テストの動作をニーズに合わせて調整することができます。 @settings デコレーターを使うことで、テストケースの最大数や各テストの実行期限、冗長性のレベルなどを制御できます。詳しくは 仮説の設定 を参照してください。

from datetime import timedelta

from hypothesis import given, settings
from snowflake.snowpark import DataFrame, Session

from snowflake.hypothesis_snowpark import dataframe_strategy


@given(
    df=dataframe_strategy(
        schema="path/to/file.json",
        session=Session.builder.getOrCreate(),
    )
)
@settings(
    deadline=timedelta(milliseconds=800),
    max_examples=25,
)
def test_my_function(df: DataFrame):
    # Test a particular function using the generated Snowpark DataFrame
    ...