SnowPro Specialty: Snowpark完全解説｜Python/Scalaデータ処理対策

Snowparkは、Python・Scala・JavaからSnowflake内のデータをプログラマブルに処理するための開発フレームワークです。 すべてのDataFrame操作はSnowflakeのウェアハウス上で実行されるため、 データをSnowflakeの外に移動させる必要がありません。 Lazy evaluationによりアクション呼び出しまでSQLは発行されず、 Snowflake Optimizerが最適な実行計画を生成します。

SnowPro Specialty: Snowpark 試験概要

出題ドメインと配点

DataFrame API主要メソッド

SnowparkのDataFrame APIはSpark風のメソッドチェーンでデータ処理を記述します。各メソッドはSQLに変換されるため、Snowflake Optimizerが最適化を行います。

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, sum as sum_, avg, count

session = Session.builder.configs(connection_params).create()

# Lazy evaluation: この時点ではSQLは実行されない
df = session.table("SALES")
result = (
    df.filter(col("SALE_DATE") >= "2026-01-01")
      .group_by("REGION", "PRODUCT_CATEGORY")
      .agg(
          sum_("AMOUNT").alias("TOTAL_SALES"),
          avg("AMOUNT").alias("AVG_SALE"),
          count("ORDER_ID").alias("ORDER_COUNT"),
      )
      .sort(col("TOTAL_SALES").desc())
)

# collect() でSQLが発行されSnowflake上で実行
rows = result.collect()
for row in rows:
    print(f"{row['REGION']}: {row['TOTAL_SALES']}")

UDF / UDTF / Stored Procedure 比較

UDF定義例（Python）

from snowflake.snowpark.functions import udf
from snowflake.snowpark.types import StringType, IntegerType

# インラインUDF
@udf(name="calculate_tax", return_type=IntegerType(),
     input_types=[IntegerType()], replace=True, is_permanent=False)
def calculate_tax(amount: int) -> int:
    return int(amount * 0.1)

# SQL呼び出し: SELECT calculate_tax(AMOUNT) FROM ORDERS;

# --- UDTF定義例 ---
from snowflake.snowpark.functions import udtf
from snowflake.snowpark.types import StructType, StructField

class SplitTags:
    def process(self, tags: str):
        for tag in tags.split(","):
            yield (tag.strip(),)

session.udtf.register(
    SplitTags,
    output_schema=StructType([StructField("TAG", StringType())]),
    input_types=[StringType()],
    name="split_tags",
    replace=True,
)
# SQL呼び出し: SELECT * FROM TABLE(split_tags('ml,ai,data'));

Stored Procedure定義例

from snowflake.snowpark.functions import sproc

@sproc(name="refresh_summary", replace=True, is_permanent=True,
       stage_location="@DEPLOY_STAGE",
       packages=["snowflake-snowpark-python"])
def refresh_summary(session: Session) -> str:
    source = session.table("RAW_EVENTS")
    summary = (
        source.group_by("EVENT_TYPE")
              .agg(count("*").alias("CNT"))
    )
    summary.write.save_as_table("EVENT_SUMMARY", mode="overwrite")
    return "EVENT_SUMMARY refreshed"

# 実行: CALL refresh_summary();

Vectorized UDF（Pandas UDF）

Vectorized UDFはpandas.Seriesを入出力に取り、行単位ではなくバッチ単位で処理するため通常のUDFより高速です。数値計算やML推論でパフォーマンスが求められる場面に適しています。

from snowflake.snowpark.functions import pandas_udf
from snowflake.snowpark.types import PandasSeriesType, IntegerType
import pandas as pd

@pandas_udf(name="batch_normalize", return_type=PandasSeriesType(IntegerType()),
            input_types=[PandasSeriesType(IntegerType())],
            replace=True)
def batch_normalize(series: pd.Series) -> pd.Series:
    mean = series.mean()
    std = series.std()
    return ((series - mean) / std * 100).astype(int)

# SQL: SELECT batch_normalize(SCORE) FROM EXAM_RESULTS;
# バッチ処理により行単位UDFの数十倍の速度

Snowpark ML

Snowpark MLはSnowflake上でMLモデルのトレーニングと推論を完結させるライブラリです。scikit-learn互換のAPIでfit/predictを実行でき、モデルはSnowflake Model Registryに登録して本番デプロイします。

from snowflake.ml.modeling.linear_model import LogisticRegression
from snowflake.ml.registry import Registry

# トレーニング
train_df = session.table("TRAINING_DATA")
model = LogisticRegression(
    input_cols=["FEATURE_A", "FEATURE_B", "FEATURE_C"],
    label_cols=["LABEL"],
    output_cols=["PREDICTION"],
)
model.fit(train_df)

# 推論
test_df = session.table("TEST_DATA")
predictions = model.predict(test_df)
predictions.write.save_as_table("PREDICTIONS", mode="overwrite")

# モデル登録
reg = Registry(session)
mv = reg.log_model(
    model,
    model_name="churn_classifier",
    version_name="v1",
    sample_input_data=train_df.limit(10),
)

Snowpark Container Services（SPCS）

SPCSはSnowflake上でDockerコンテナを実行するマネージドサービスです。Snowparkが主にDataFrame操作とUDF/Sprocの実行を担うのに対し、SPCSはGPUベースのモデルトレーニング、カスタムREST API、フルスタックWebアプリケーションなどの汎用ワークロードを実行できます。

学習ロードマップ

サンプル問題

よくある質問