🦔

Snowflake ML関数とTROCCOでMLモデルを作って動かす

2025/12/23に公開

はじめに

本記事は TROCCO&COMETA Advent Calendar 2025 17日目の記事です。
※2025/12/23に公開

Snowflake ML関数を用いたモデルの作成とTROCCOによるワークフローの運用で以下を試みました。

AutoML製品を使用しないで簡単にMLモデルを作る
MLモデルの運用フローをブラックボックス化させない

この記事ではサンプルデータセットSNOWFLAKE_SAMPLE_DATA.TPCH_SF1を用いてサプライチェーンにおける配送遅延リスクの予測を例としています。

サンプルデータの用意

SNOWFLAKE_SAMPLE_DATA.TPCH_SF1からLINEITEMとORDERSを用いてサンプルデータを作成します。

SQL

CREATE OR REPLACE TABLE SHIPMENT_DATA AS
SELECT
  L_ORDERKEY
  , L_LINENUMBER
  , L_QUANTITY
  , L_EXTENDEDPRICE
  , L_DISCOUNT
  , L_SHIPMODE
  , O_ORDERPRIORITY
  , O_TOTALPRICE
  , O_ORDERDATE
  , L_SHIPDATE
  , L_COMMITDATE
  -- ターゲット変数:予定日(COMMITDATE)より出荷日(SHIPDATE)が遅れたら1
  , CASE WHEN L_SHIPDATE > L_COMMITDATE THEN 1 ELSE 0 END AS IS_DELAYED
FROM
  SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.LINEITEM
INNER JOIN
  SNOWFLAKE_SAMPLE_DATA.TPCH_SF1.ORDERS
  ON
    L_ORDERKEY = O_ORDERKEY
LIMIT 10000;

L_ORDERKEY	L_LINENUMBER	L_QUANTITY	L_EXTENDEDPRICE	L_DISCOUNT	L_SHIPMODE	O_ORDERPRIORITY	O_TOTALPRICE	O_ORDERDATE	L_SHIPDATE	L_COMMITDATE	IS_DELAYED
4200001	1	35	36718.85	0.04	REG AIR	3-MEDIUM	99406.41	1994-02-21	1994-06-06	1994-04-20	1
4200001	2	32	61499.84	0.05	AIR	3-MEDIUM	99406.41	1994-02-21	1994-05-05	1994-04-28	1
4200002	1	6	8029.86	0.05	MAIL	4-NOT SPECIFIED	256838.41	1997-04-14	1997-07-21	1997-06-21	1
4200002	2	41	43380.05	0.06	SHIP	4-NOT SPECIFIED	256838.41	1997-04-14	1997-05-18	1997-06-17	0
4200002	3	48	74141.76	0.05	AIR	4-NOT SPECIFIED	256838.41	1997-04-14	1997-07-30	1997-06-08	1

特徴量ストアの作成

L_ORDERKEYとL_LINENUMBERから一意のエンティティを定め、それに関連する属性を特徴量ビューとして登録します。

エンティティの作成

Python

from snowflake.ml.feature_store import FeatureStore, Entity, FeatureView
from snowflake.snowpark import Session
import snowflake.snowpark.functions as F

session = get_active_session()

# 特徴量ストアを定義
fs = FeatureStore(
    session=session, 
    database="MLOPS", 
    name="FEATURE_STORE", 
    default_warehouse="DEFAULT_WH_S"
)

# エンティティを定義
# 一意となるキーとしてL_ORDERKEYとL_LINENUMBERの組み合わせを用いる
order_item_entity = Entity(name="ORDER_ITEM", join_keys=["L_ORDERKEY", "L_LINENUMBER"])
# エンティティを登録
fs.register_entity(order_item_entity)

特徴量ビューの作成

Python

# 特徴量ビューをSQLで定義
fv_query = """
SELECT 
    L_ORDERKEY
  , L_LINENUMBER
  , L_QUANTITY
  , L_EXTENDEDPRICE
  , L_DISCOUNT
  , L_SHIPMODE
  , O_ORDERPRIORITY
  , O_TOTALPRICE
  , O_ORDERDATE
  , L_SHIPDATE
  , L_COMMITDATE
  -- 追加する特徴量 --
  -- 計画リードタイム
  , DATEDIFF('day', O_ORDERDATE, L_COMMITDATE) AS EXPECTED_LEAD_TIME
  -- 1注文あたりの品目数
  , COUNT(*) OVER (PARTITION BY L_ORDERKEY) AS ITEMS_IN_ORDER
  -- 1個あたりの単価
  , L_EXTENDEDPRICE / NULLIF(L_QUANTITY, 0) AS UNIT_PRICE
FROM MLOPS.EXSAMPLE.SHIPMENT_DATA
"""
shipment_fv = FeatureView(
    name="SHIPMENT_FEATURES", 
    entities=[order_item_entity], 
    feature_df=session.sql(fv_query)
)

# 特徴量ビューを登録
shipment_fv = fs.register_feature_view(feature_view=shipment_fv, version="V1")

学習用データセットの作成

特徴量とターゲットを結合して学習用スナップショットを保存します。

Python

# Spine(特徴量ビューから特徴量を取得するためのキーを備えたデータフレーム)となるデータの準備
# 結合キー（L_ORDERKEYとL_LINENUMBER）、ターゲット（IS_DELAYED）を選択
spine_df = session.table("MLOPS.EXSAMPLE.SHIPMENT_DATA").select(
    "L_ORDERKEY", "L_LINENUMBER", "IS_DELAYED")

# データセットの生成
# 特徴量ビューとラベルデータを結合し、Snowflake上のDatasetとして保存
training_dataset = fs.generate_dataset(
    name="DELIVERY_DELAY_TRAINING_DATASET",
    spine_df=spine_df,
    features=[shipment_fv],  # 先ほど作成した特徴量ビューを指定
    version="V1",
    output_type="dataset"
)

MLモデルの作成

Snowflake ML関数を使って学習を行います。

ML関数でデータセットを利用するためにテーブルを作成

Python

from snowflake.ml.dataset import load_dataset

# データセットを読み込む
ds = load_dataset(
    session=session,
    name="MLOPS.FEATURE_STORE.DELIVERY_DELAY_TRAINING_DATASET",
    version="V1"
)

# データセットをSnowpark DataFrameとして取得
training_df = ds.read.to_snowpark_dataframe().select(
    "L_ORDERKEY",
    "L_LINENUMBER",
    "IS_DELAYED",
    "L_QUANTITY",
    "L_EXTENDEDPRICE",
    "L_DISCOUNT",
    "L_SHIPMODE",
    "O_ORDERPRIORITY",
    "O_TOTALPRICE",
    # ML関数利用時は日付をTIMESTAMP_NTZに変換する処理が必要
    F.col("O_ORDERDATE").cast("TIMESTAMP_NTZ").alias("O_ORDER_TIMESTAMP"),
    "EXPECTED_LEAD_TIME",
    "ITEMS_IN_ORDER",
    "UNIT_PRICE"
)

# ML関数で参照可能なテーブルとして保存
training_df.write.mode("overwrite").save_as_table("MLOPS.FEATURE_STORE.DELIVERY_DELAY_TRAINING")

SNOWFLAKE.ML.CLASSIFICATION関数でモデルをトレーニング

SQL

CREATE OR REPLACE SNOWFLAKE.ML.CLASSIFICATION DELAY_CLASSIFIER(
    INPUT_DATA => TABLE(MLOPS.FEATURE_STORE.DELIVERY_DELAY_TRAINING),
    TARGET_COLNAME => 'IS_DELAYED'
);

指標の確認

用意されているメソッドで特徴量の重要性や評価指標を確認します。

特徴量重要度 SHOW_FEATURE_IMPORTANCE

SQL

CALL DELAY_CLASSIFIER!SHOW_FEATURE_IMPORTANCE();

RANK	FEATURE	SCORE	FEATURE_TYPE
1	UNIT_PRICE	0.1306666667	user_provided
2	O_TOTALPRICE	0.109	user_provided
3	EXPECTED_LEAD_TIME	0.1066666667	user_provided
4	L_EXTENDEDPRICE	0.09633333333	user_provided
5	O_ORDER_TIMESTAMP_epoch_time	0.09633333333	derived_feature
6	O_ORDER_TIMESTAMP_day_of_year	0.08	derived_feature
7	L_QUANTITY	0.072	user_provided
8	O_ORDER_TIMESTAMP_day_of_quarter	0.067	derived_feature
9	O_ORDER_TIMESTAMP_day	0.06366666667	derived_feature
10	L_DISCOUNT	0.059	user_provided

評価指標 SHOW_GLOBAL_EVALUATION_METRICS

SQL

CALL DELAY_CLASSIFIER!SHOW_GLOBAL_EVALUATION_METRICS();

DATASET_TYPE	AVERAGE_TYPE	ERROR_METRIC	METRIC_VALUE
EVAL	macro	precision	0.6310582758
EVAL	macro	recall	0.6309836085
EVAL	macro	f1	0.6302128416
EVAL	macro	auc	0.6739029804
EVAL	weighted	precision	0.6317299462
EVAL	weighted	recall	0.6302186879
EVAL	weighted	f1	0.6301660718
EVAL	weighted	auc	0.6739029804
EVAL		log_loss	0.655435834

混同行列 SHOW_CONFUSION_MATRIX

SQL

CALL DELAY_CLASSIFIER!SHOW_CONFUSION_MATRIX();

DATASET_TYPE	ACTUAL_CLASS	PREDICTED_CLASS	COUNT
EVAL	0	0	319
EVAL	0	1	168
EVAL	1	0	204
EVAL	1	1	315

精度は良くありませんが今回はこのまま作成したモデルを使用します。

SQL上での予測の実行

PREDICTメソッドで実行した予測結果をテーブルに保存して渡したいデータを出力します。

予測結果をテーブルに保存

SQL

CREATE OR REPLACE TABLE MLOPS.EXSAMPLE.DELIVERY_DELAY_FORCASTED_TEST AS
SELECT
    L_ORDERKEY
    , L_LINENUMBER
    , L_SHIPMODE
    , O_ORDER_TIMESTAMP::DATE AS O_ORDER_DATE
    -- PREDICTメソッドで予測を実行
    , DELAY_CLASSIFIER!PREDICT(
        INPUT_DATA => OBJECT_CONSTRUCT(
            'L_QUANTITY', L_QUANTITY
            , 'L_EXTENDEDPRICE', L_EXTENDEDPRICE
            , 'L_DISCOUNT', L_DISCOUNT
            , 'L_SHIPMODE', L_SHIPMODE
            , 'O_ORDERPRIORITY', O_ORDERPRIORITY
            , 'O_TOTALPRICE', O_TOTALPRICE
            , 'O_ORDER_TIMESTAMP', O_ORDER_TIMESTAMP
            , 'EXPECTED_LEAD_TIME', EXPECTED_LEAD_TIME
            , 'ITEMS_IN_ORDER', ITEMS_IN_ORDER
            , 'UNIT_PRICE', UNIT_PRICE
        )
    ) AS PREDICTION
FROM
  MLOPS.FEATURE_STORE.DELIVERY_DELAY_TRAINING;

整形した予測結果の出力

SQL

SELECT
  L_ORDERKEY
  , L_LINENUMBER
  , L_SHIPMODE
  , O_ORDER_DATE
  -- 遅延する確率を取得
  , ROUND(PREDICTION:probability:"1",4) AS DELAY_PROBABILITY
FROM
  MLOPS.EXSAMPLE.DELIVERY_DELAY_FORCASTED_TEST
LIMIT 5;

L_ORDERKEY	L_LINENUMBER	L_SHIPMODE	O_ORDER_DATE	DELAY_PROBABILITY
4200001	2	AIR	1994-02-21	0.4694
4200003	1	REG AIR	1997-11-24	0.4872
4200005	2	TRUCK	1997-02-01	0.5954
4200006	4	MAIL	1993-12-31	0.1272
4200033	4	RAIL	1996-03-24	0.2491

L_ORDERKEY×L_LINENUMBER単位の配送遅延

TROCCOを用いたワークフローの作成

前日に受注したオーダーの配送遅延リスクを予測するフローを作成します。
TROCCOを用いて以下の処理をスケジュール実行します。

予測用テーブルの作成
予測の実行
予測結果を出力してSFTPで送信

接続情報の作成

公式ドキュメントを参照してTROCCOのSnowflake接続情報を作成します。
（例: Windows11の場合）

秘密鍵と公開鍵の生成と認証の設定

Git Bashで秘密鍵を生成します。

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt

続いて公開鍵を生成します。

openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub

公開鍵を整形します。

cat rsa_key.pub | \
tr -d "\n" | \
sed "s/-----BEGIN PUBLIC KEY-----//g" | \
sed "s/-----END PUBLIC KEY-----//g"

SnowflakeでTROCCO接続用のサービスユーザーを作成します。

SQL

CREATE USER TROCCO_CONNECTOR
  TYPE = SERVICE
  RSA_PUBLIC_KEY = '{PUBLIC_KEY}'
  MUST_CHANGE_PASSWORD = FALSE;

ロールへの権限付与

SQL

USE ROLE ACCOUNTADMIN;
USE DATABASE MLOPS;
USE SCHEMA MLOPS.EXSAMPLE;

-- 作成したユーザーに割り当てるためのロールを作成する
CREATE ROLE SERVICE_ROLE;
GRANT ROLE SERVICE_ROLE TO ROLE ACCOUNTADMIN;

-- ウェアハウスの利用権限を付与
GRANT USAGE ON WAREHOUSE DEFAULT_WH_S TO ROLE SERVICE_ROLE;
GRANT OPERATE ON WAREHOUSE DEFAULT_WH_S TO ROLE SERVICE_ROLE;
-- アクセス対象DBに対するUSAGE権限を付与
GRANT USAGE ON DATABASE MLOPS TO ROLE SERVICE_ROLE;
-- アクセス対象DBの既存の全てのスキーマに対するUSAGE権限を付与
GRANT USAGE ON ALL SCHEMAS IN DATABASE MLOPS TO ROLE SERVICE_ROLE;
-- アクセス対象DBの今後作成される全てのスキーマに対するUSAGE権限を付与
GRANT USAGE ON FUTURE SCHEMAS IN DATABASE MLOPS TO ROLE SERVICE_ROLE;
-- アクセス対象DBのテーブル作成権限を付与
GRANT CREATE TABLE ON SCHEMA MLOPS.EXSAMPLE TO ROLE SERVICE_ROLE;
-- 作成したモデルのPREDICTメソッド呼び出し権限を付与
GRANT SNOWFLAKE.ML.CLASSIFICATION ROLE DELAY_CLASSIFIER!MLCONSUMER TO ROLE SERVICE_ROLE;
-- 使用する既存テーブルの所有権を付与
GRANT OWNERSHIP ON TABLE MLOPS.EXSAMPLE.SHIPMENT_DATA TO ROLE SERVICE_ROLE COPY CURRENT GRANTS;

TROCCO Snowflake接続情報の作成

TROCCOで新規のSnowflake接続情報を作成します。
ユーザーと秘密鍵、ロールに上記で作成したものを入力します。

ワークフローの作成

TROCCOのデータ転送やデータマート用いてワークフローを作ります。

予測用データを作成するデータマート

受注日が前日のレコードを抽出して予測用データとします。
DELIVERY_DELAY_PREDICTION_INPUTというテーブルを作成します。

SQL

SELECT 
    L_ORDERKEY
  , L_LINENUMBER
  , L_QUANTITY
  , L_EXTENDEDPRICE
  , L_DISCOUNT
  , L_SHIPMODE
  , O_ORDERPRIORITY
  , O_TOTALPRICE
  , CAST(O_ORDERDATE AS TIMESTAMP_NTZ) AS O_ORDER_TIMESTAMP
  , L_SHIPDATE
  , L_COMMITDATE
  , DATEDIFF('day', O_ORDERDATE, L_COMMITDATE) AS EXPECTED_LEAD_TIME
  , COUNT(*) OVER (PARTITION BY L_ORDERKEY) AS ITEMS_IN_ORDER
  , L_EXTENDEDPRICE / NULLIF(L_QUANTITY, 0) AS UNIT_PRICE
FROM
  MLOPS.EXSAMPLE.SHIPMENT_DATA
WHERE
  O_ORDERDATE::DATE = DATEADD(day, -1, CURRENT_DATE())

予測結果を出力するデータマート

モデルを呼び出して予測結果をテーブルとして生成します。
DELIVERY_DELAY_FORCASTEDというテーブルを作成します。

SQL

SELECT
    L_ORDERKEY
    , L_LINENUMBER
    , L_SHIPMODE
    , O_ORDER_TIMESTAMP::DATE AS O_ORDER_DATE
    -- 遅延する確率を取得
    , MLOPS.EXSAMPLE.DELAY_CLASSIFIER!PREDICT(
        INPUT_DATA => OBJECT_CONSTRUCT(
            'L_QUANTITY', L_QUANTITY
            , 'L_EXTENDEDPRICE', L_EXTENDEDPRICE
            , 'L_DISCOUNT', L_DISCOUNT
            , 'L_SHIPMODE', L_SHIPMODE
            , 'O_ORDERPRIORITY', O_ORDERPRIORITY
            , 'O_TOTALPRICE', O_TOTALPRICE
            , 'O_ORDER_TIMESTAMP', O_ORDER_TIMESTAMP
            , 'EXPECTED_LEAD_TIME', EXPECTED_LEAD_TIME
            , 'ITEMS_IN_ORDER', ITEMS_IN_ORDER
            , 'UNIT_PRICE', UNIT_PRICE
        )
    ) AS PREDICTION
FROM
  MLOPS.EXSAMPLE.DELIVERY_DELAY_PREDICTION_INPUT

SnowflakeからSFTPでサーバーにCSVを送る転送設定

予測結果(配送遅延確率)をCSVに出力してデータを利用するサーバーにアップロードします。

SQL

SELECT
  L_ORDERKEY
  , L_LINENUMBER
  , L_SHIPMODE
  , ROUND(PREDICTION:probability:"1",4) AS DELAY_PROBABILITY
FROM
  MLOPS.EXSAMPLE.DELIVERY_DELAY_FORCASTED

ワークフローの作成

上記をワークフローとしてまとめてスケジュール実行します。
依存関係もひと目でわかります。

作成したワークフロー

これで毎日MLモデルで予測を自動実行して結果を送信することができるようになりました。

おわりに

ML関数はアルゴリズムの選択や細かいチューニングができなかったり、モデルレジストリに登録されなかったりするなどのデメリットはありますが、SQLを実行するだけで特徴量の派生特徴量の追加やハイパーパラメータの調整を自動で行ってモデルを作成してくれます。
また、TROCCOを用いることによって、運用においてもほぼSQLのみでMLパイプラインの自動化を行うと共に誰でもフローの構造や依存関係が把握できるようになります。
データアナリストがSQLのみで機械学習を行える環境を用意する必要がある場合、この組み合わせはひとつの手段として考えられるかもしれません。

はじめに

サンプルデータの用意

特徴量ストアの作成

学習用データセットの作成

MLモデルの作成

指標の確認

SQL上での予測の実行

TROCCOを用いたワークフローの作成

接続情報の作成

ワークフローの作成

おわりに

Discussion