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

Snowpark変換コードの検証¶

Snowparkチェックポイントパッケージは、Snowparkコードに適用できる検証セットを提供し、 PySpark コードとの動作の等価性を保証します。

フレームワークが提供する関数¶

check_with_spark：Snowpark DataFrame の引数を関数あるいはサンプルに変換し、 PySpark DataFrames に変換するデコレーター。このチェックでは、新しいSnowpark関数の機能を反映したspark関数がプロバイダーから提供されて実行され、2つの実装間の出力が比較されます。Spark関数とSnowpark関数が意味的に同一であると仮定して、実際のサンプルデータ上でこれらの関数を検証します。
パラメーター：
- job_context (SnowparkJobContext)：検証の構成と詳細を含むジョブコンテキスト。
- spark_function (fn)：Snowparkの実装と比較するための同等の PySpark 関数。
- checkpoint_name (str）：チェックポイントの名前。デフォルトは None。
- sample_number (Optional[int],　オプション)：検証の行数。デフォルトは100。
- sampling_strategy (Optional[SamplingStrategy], オプション)：データのサンプリングに使用するストラテジー。デフォルトは SamplingStrategy.RANDOM_SAMPLE です。
- output_path (Optional[str],　オプション)：検証結果を保存するパス。デフォルトは None。
以下はその例です：
```
 def original_spark_code_I_dont_understand(df):
 from pyspark.sql.functions import col, when

 ret = df.withColumn(
     "life_stage",
     when(col("byte") < 4, "child")
     .when(col("byte").between(4, 10), "teenager")
     .otherwise("adult"),
 )
 return ret


@check_with_spark(
 job_context=job_context, spark_function=original_spark_code_I_dont_understand
)
def new_snowpark_code_I_do_understand(df):
  from snowflake.snowpark.functions import col, lit, when

  ref = df.with_column(
      "life_stage",
      when(col("byte") < 4, lit("child"))
      .when(col("byte").between(4, 10), lit("teenager"))
      .otherwise(lit("adult")),
 )
 return ref


 df1 = new_snowpark_code_I_do_understand(df)
```
Copy
validate_dataframe_checkpoint：この関数はSnowparkデータフレームを引数モードに従って、特定のチェックポイントスキーマファイルまたはインポートされたデータフレームに対して検証します。これは、その DataFrame、関数に渡される DataFrame、収集された情報が同等であることを保証します。
パラメーター：
df (SnowparkDataFrame)： DataFrame を検証します。

checkpoint_name (str)：検証するチェックポイントの名前。

job_context (SnowparkJobContext, オプション) (str)：検証のジョブコンテキスト。PARQUET モードに必要です。

mode (CheckpointMode)：検証モード（例： SCHEMA、 PARQUET）デフォルトは SCHEMA です。

custom_checks (オプション[dict[Any, Any]]、オプション）：検証中に適用するカスタムチェック。

skip_checks (オプション[dict[Any, Any]]、オプション）：検証中にスキップするチェック。

sample_frac (Optional[float], オプション)：検証用にサンプルする DataFrame の端数。デフォルトは0.1。

sample_number (Optional[int]　オプション)：検証用にサンプルする行の数。

sampling_strategy (Optional[SamplingStrategy], オプション)：サンプリングに使用する戦略。

output_path (Optional[str],　オプション)：検証結果の出力パス。
以下はその例です：
# Check a schema/stats here! validate_dataframe_checkpoint( df1, "demo_add_a_column_dataframe", job_context=job_context, mode=CheckpointMode.DATAFRAME, # CheckpointMode.Schema) )
Copy
選択されたモードに応じて、検証は収集されたスキーマファイルまたはSnowflakeのParquetロードされたDataframeのいずれかを使用して、 PySpark バージョンとの依存関係を検証します。

check-output_schema: このデコレータはSnowpark関数の出力のスキーマを検証し、出力 DataFrame が指定されたPanderaスキーマに準拠していることを保証します。特にSnowparkパイプラインでデータの統合と一貫性を強制するのに便利です。このデコレータは、検証対象の Pandera スキーマ、チェックポイント名、サンプリングパラメーター、そしてオプションのジョブコンテキストなどのパラメーターを受け取ります。Snowpark関数をラップし、結果を返す前に出力 DataFrame に対してスキーマ検証を行います。

以下はその例です：

from pandas import DataFrame as PandasDataFrame
from pandera import DataFrameSchema, Column, Check
from snowflake.snowpark import Session
from snowflake.snowpark import DataFrame as SnowparkDataFrame
from snowflake.snowpark_checkpoints.checkpoint import check_output_schema
from numpy import int8

# Define the Pandera schema
out_schema = DataFrameSchema(
{
    "COLUMN1": Column(int8, Check.between(0, 10, include_max=True, include_min=True)),
    "COLUMN2": Column(float, Check.less_than_or_equal_to(-1.2)),
    "COLUMN3": Column(float, Check.less_than(10)),
}
)

# Define the Snowpark function and apply the decorator
@check_output_schema(out_schema, "output_schema_checkpoint")
def preprocessor(dataframe: SnowparkDataFrame):
 return dataframe.with_column(
    "COLUMN3", dataframe["COLUMN1"] + dataframe["COLUMN2"]
)

# Create a Snowpark session and DataFrame
session = Session.builder.getOrCreate()
df = PandasDataFrame(
{
    "COLUMN1": [1, 4, 0, 10, 9],
    "COLUMN2": [-1.3, -1.4, -2.9, -10.1, -20.4],
}
)

sp_dataframe = session.create_dataframe(df)

# Apply the preprocessor function
preprocessed_dataframe = preprocessor(sp_dataframe)

Copy

check_input_schema：このデコレータはSnowpark関数の入力引数のスキーマを検証します。このデコレータは、関数が実行される前に入力 DataFrame が指定した Pandera スキーマに準拠していることを確認します。特にSnowparkパイプラインでデータの統合と一貫性を強制するのに便利です。このデコレータは、検証対象の Pandera スキーマ、チェックポイント名、サンプリングパラメーター、そしてオプションのジョブコンテキストなどのパラメーターを受け取ります。Snowpark関数をラップし、関数を実行する前に入力 DataFrame に対してスキーマ検証を行います。

以下はその例です：

from pandas import DataFrame as PandasDataFrame
from pandera import DataFrameSchema, Column, Check
from snowflake.snowpark import Session
from snowflake.snowpark import DataFrame as SnowparkDataFrame
from snowflake.snowpark_checkpoints.checkpoint import check_input_schema
from numpy import int8

# Define the Pandera schema
input_schema = DataFrameSchema(
{
    "COLUMN1": Column(int8, Check.between(0, 10, include_max=True, include_min=True)),
    "COLUMN2": Column(float, Check.less_than_or_equal_to(-1.2)),
}
)

# Define the Snowpark function and apply the decorator
@check_input_schema(input_schema, "input_schema_checkpoint")
def process_dataframe(dataframe: SnowparkDataFrame):
return dataframe.with_column(
    "COLUMN3", dataframe["COLUMN1"] + dataframe["COLUMN2"]
)

# Create a Snowpark session and DataFrame
session = Session.builder.getOrCreate()
df = PandasDataFrame(
{
    "COLUMN1": [1, 4, 0, 10, 9],
    "COLUMN2": [-1.3, -1.4, -2.9, -10.1, -20.4],
}
)
sp_dataframe = session.create_dataframe(df)

# Apply the process_dataframe function
processed_dataframe = process_dataframe(sp_dataframe)

Copy

統計チェック¶

Schema モードで検証を実行すると、デフォルトで特定の列タイプに対して統計情報の検証が適用されます。 skip_checks でこれらのチェックをスキップすることができます。

列タイプ	デフォルトチェック
数値： `byte`, `short`, `integer`, `long`, `float`, および `double`	間：値が最小値と最大値の間にある場合、最小値と最大値を含みます。 decimal_precision：値が10進数の場合、10進数の精度をチェックします。平均：列の平均が特定の範囲内にあるかどうかを検証します。
ブール値	isin：値が True か False かを検証します。 True_proportion：True 値の割合が特定の範囲内にあるかどうかを検証します。 False_proportion：False 値の割合が特定の範囲内にあるかどうかの検証。
日付： `date`, `timestamp`, そして `timestamp_ntz`	間：値が最小値と最大値の間にある場合、最小値と最大値を含みます。
Nullable：サポートされるすべてのタイプ	Null_proportion：nullプロポーションを検証します。

スキップ・チェック¶

チェックのための細粒度の制御では、列の検証や特定の列のチェックをスキップすることができます。 skip_checks というパラメーターで、スキップしたい列とバリデーションタイプを指定できます。スキップに使用されるチェックの名前は、そのチェックに関連付けられているものです。

str_contains
str_endswith
str_length
str_matches
str_startswith
in_range
equal_to
greater_than_or_equal_to
greater_than
less_than_or_equal_to
less_than
not_equal_to
notin
isin

df = pd.DataFrame(
{
      "COLUMN1": [1, 4, 0, 10, 9],
      "COLUMN2": [-1.3, -1.4, -2.9, -10.1, -20.4],
}
)

schema = DataFrameSchema(
{
      "COLUMN1": Column(int8, Check.between(0, 10, element_wise=True)),
      "COLUMN2": Column(
          float,
          [
              Check.greater_than(-20.5),
              Check.less_than(-1.0),
              Check(lambda x: x < -1.2),
          ],
      ),
}
)

session = Session.builder.getOrCreate()
sp_df = session.create_dataframe(df)
check_dataframe_schema(
  sp_df,
  schema,
  skip_checks={"COLUMN1": [SKIP_ALL], "COLUMN2": ["greater_than", "less_than"]},
)

Copy

カスタムチェック¶

custom_checks プロパティを使って、 JSON ファイルから生成されたスキーマに追加のチェックを加えることができます。これでチェックがPandera・スキーマに追加されます。

df = pd.DataFrame(
  {
        "COLUMN1": [1, 4, 0, 10, 9],
        "COLUMN2": [-1.3, -1.4, -2.9, -10.1, -20.4],
  }
)

session = Session.builder.getOrCreate()
sp_df = session.create_dataframe(df)

# Those check will be added to the schema generate from the JSON file
result = validate_dataframe_checkpoint(
  sp_df,
  "checkpoint-name",
  custom_checks={
        "COLUMN1": [
            Check(lambda x: x.shape[0] == 5),
            Check(lambda x: x.shape[1] == 2),
    ],
    "COLUMN2": [Check(lambda x: x.shape[0] == 5)],
  },
)

Copy

サンプル戦略¶

プロバイダー・コードのサンプリング・プロセスは、データの代表サンプルを取ることによって、大容量 DataFrames を効率的に検証するように設計されています。このアプローチは、計算コストと時間のかかるデータセット全体を処理することなくスキーマ検証を行うのに役立ちます。

パラメーター：

sample_frac：このパラメーターは、サンプルする DataFrame の割合を指定します。例えば、 sample_frac を0.1にセットすると、 DataFrame の行の10％がサンプルされます。これは、計算リソースを節約するためにデータのサブセットを検証したい場合に便利です。
sample_number: このパラメーターは、 DataFrame からサンプルする正確な行数を指定します。例えば、 sample_number を100にセットすると、 DataFrame から100行がサンプルされます。これは、 DataFrame のサイズに関係なく、固定行数を検証したい場合に便利です。

検証結果¶

どのタイプであれ、検証が実行されると、その結果は合格であれ不合格であれ、 checkpoint_validation_results.json に保存されます。このファイルは主に、 VSCode 拡張子の関数に使用されます。検証のステータス、タイムスタンプ、チェックポイント名、関数の実行が行われた行数、ファイルに関する情報が含まれます。

また、検証結果はデフォルトのSnowflakeアカウントにログとして記録され、 SNOWPARK_CHECKPOINTS_REPORT というテーブルに検証結果に関する情報が格納されます。

DATE: 検証の実行タイムスタンプ。
JOB： SnowparkJobContext の名前。
STATUS: 検証のステータス。
CHECKPOINT: 検証されたチェックポイントの名前。
MESSAGE: エラーメッセージ
DATA: 検証実行時のデータ。
EXECUTION_MODE:　検証モード実行。