Snowpark Container Servicesのモデルサービス¶

コンピューティングプールの作成¶

Snowpark Container Services （SPCS） は、コンピューティングプールでコンテナーイメージを実行します。適切なコンピューティングプールがまだない場合は、以下のようにして作成します。

CREATE COMPUTE POOL IF NOT EXISTS mypool
    MIN_NODES = 2
    MAX_NODES = 4
    INSTANCE_FAMILY = 'CPU_X64_M'
    AUTO_RESUME = TRUE;

有効なインスタンスファミリーの一覧は、 ファミリ名のテーブル をご参照ください。

モデルを実行するロールが、コンピューティングプールのオーナーであるか、あるいは、 USAGE 、あるいは、 OPERATE 、そのプールの権限を持っていることをご確認しださい。

モデルの依存性と適格性¶

モデルの依存関係は、それがウェアハウスで実行できるか、 SPCS サービスで実行できるか、あるいはその両方で実行できるかを決定します。必要であれば、依存関係を意図的に指定して、モデルをこれらの環境のいずれかで実行できないようにすることができます。

Snowflake の conda チャネルはウェアハウスでのみ利用可能で、ウェアハウスの依存関係の唯一のソースです。デフォルトでは、 SPCS モデルの conda 依存関係は conda-forge から取得します。

モデルのバージョンをログに記録すると、モデルの conda 依存関係が Snowflake の conda チャンネルに対して検証されます。モデルのすべてのconda依存関係がそこで利用可能であれば、そのモデルはウェアハウスで実行する資格があるとみなされます。また、依存関係がすべてconda-forgeから利用可能であれば、 SPCS サービスで実行することもできます。ただし、これはサービスを作成するまで確認されません。

PyPI の依存関係でログに記録されたモデルは、 SPCS で実行する必要があります。少なくとも1つの PyPI 依存関係を指定することは、モデルをウェアハウスで実行する資格がないようにする1つの方法です。モデルにcondaの依存関係しかない場合は、次の例に示すように、少なくとも1つのチャネルを明示的に指定します（conda-forgeでも構いません）。

# reg is a snowflake.ml.registry.Registry object
reg.log_model(
    model_name="my_model",
    version_name="v1",
    model=model,
    conda_dependencies=["conda-forge::scikit-learn"])

SPCS-deployedモデルの場合、condaの依存関係があれば、まずインストールされ、次に PyPI の依存関係があれば、 pip を使ってconda環境にインストールされます。

サービスの作成¶

SPCS サービスを作成し、そこにモデルをデプロイするには、次の例に示すように、モデルバージョンの create_service メソッドを呼び出します。

# mv is a snowflake.ml.model.ModelVersion object
mv.create_service(service_name="myservice",
                  service_compute_pool="my_compute_pool",
                  ingress_enabled=True,
                  gpu_requests=None)

create_service に必要な引数を以下に示します。

• service_name: 作成するサービス名。この名前はアカウント内で一意である必要があります。

• service_compute_pool: モデルの実行に使用するコンピュートプールの名前。コンピューティングプールはすでに存在している必要があります。

• ingress_enabled:True の場合、サービスは HTTP エンドポイント経由でアクセスできるようになります。エンドポイントを作成するには、 BIND SERVICE ENDPOINT の権限が必要です。

• gpu_requests: GPUs の数を指定する文字列。CPU または GPU のいずれかで実行できるモデルの場合、この引数は、モデルが CPU と GPUs のどちらで実行されるかを決定します。モデルが CPU （例えばscikit-learnモデル）でしか実行できない既知のタイプの場合、 GPUs を要求するとイメージビルドは失敗します。

新しいモデルをデプロイする場合、CPU駆動型モデルのサービスを作成するのに最大5分、GPU駆動型モデルを作成するのに10分かかります。コンピューティングプールがアイドル状態であるか、サイズ変更が必要な場合は、サービスの作成に時間がかかる可能性があります。

This example shows only the required and most commonly used arguments. See the ModelVersion API reference for a complete list of arguments.

デフォルトのサービス構成¶

Inference Serverは、ほとんどの大文字と小文字で動作する使いやすいデフォルトを使用しています。そのセットは以下です。

• ワーカースレッド数： CPU-poweredモデルの場合、サーバーは CPUs の2倍＋1個のワーカープロセスを使用します。GPU使用型はワーカープロセスを1つ使います。create_service の呼び出しでは、 num_workers 引数を使用してオーバーライドできます。

• スレッドセーフティ：スレッドセーフでないモデルもあります。そのため、サービスはワーカープロセスごとにモデルの別個のコピーをロードします。その結果、大規模なモデルではリソースが枯渇する可能性があります。

• ノード利用：デフォルトでは、1つの推論サーバーインスタンスは、実行するノードのすべての CPU とメモリをリクエストすることで、ノード全体をリクエストします。インスタンスごとのリソース割り当てをカスタマイズするには、 cpu_requests、 memory_requests、 gpu_requests のような引数を使用します。

• エンドポイント：推論エンドポイントは inference と名付けられ、ポート5000を使用します。これらはカスタマイズできません。

• Auto Suspend:推論サービスは、非アクティブな状態が30分経過すると、デフォルトで自動的に一時停止します。一時停止したサービスは、リクエストを受け取ると自動的に再開されます。

コンテナイメージのビルド動作¶

デフォルトでは、Snowflake Model Servingは、モデルの実行に使用されるのと同じコンピューティングプールを使用してコンテナーイメージを構築します。この推論コンピューティングプールは、このタスクにはオーバーパワーである可能性が高いです（例えば、 GPUs は、コンテナーイメージの構築には使用されません）。ほとんどの場合、これはコンピューティングコストに大きな影響を与えませんが、気になる場合は、 image_build_compute_pool 引数を指定することでイメージのビルドのためによりパワフルではないコンピューティングプールを選択できます。

create_service はべき等関数です。複数回呼び出しても、毎回イメージ構築のトリガーにはなりません。しかし、コンテナーイメージは、依存パッケージの脆弱性の修正を含む推論サービスの更新に基づいて再構築される可能性があります。この場合、 create_service 、自動的にイメージの再構築がトリガーされます。

ユーザーインターフェイス¶

展開されたモデルは、Model Registry Snowsight UI で管理できます。詳細については、 モデル推論サービス をご参照ください。

SQL¶

Snowflake Model Serving は、モデルを SPCS にデプロイする際にサービス関数を作成します。これらの関数は、 SQL から、 SPCS のコンピューティングプールで実行されているモデルへのブリッジとして機能します。モデルの各メソッドに対して1つのサービス関数が作成され、それらは service_name!method_name のように命名されます。例えば、モデルが PREDICT と EXPLAIN という2つのメソッドを持ち、 MY_SERVICE というサービスにデプロイされる場合、結果として生じるサービス関数は MY_SERVICE!PREDICT と MY_SERVICE!EXPLAIN となります。

SQL のモデルのサービス関数を呼び出すには、以下のようなコードを使用します。

-- See signature of the inference function in SQL.
SHOW FUNCTIONS IN SERVICE MY_SERVICE;

-- Call the inference function in SQL following the same signature (from `arguments` column of the above query)
SELECT MY_SERVICE!PREDICT(feature1, feature2, ...) FROM input_data_table;

Python¶

モデルバージョンオブジェクトの run メソッドを使用して、サービスのメソッドを呼び出します。service_name 引数を指定して、メソッドが実行されるサービスを指定することを含みます。例:

# Get signature of the inference function in Python
# mv: snowflake.ml.model.ModelVersion
mv.show_functions()
# Call the function in Python
service_prediction = mv.run(
    test_df,
    function_name="predict",
    service_name="my_service")

service_name 引数を含めない場合、モデルはウェアハウスで実行されます。

HTTP endpoints¶

Every service comes with its internal DNS name. Deploying a service with ingress_enabled also creates a public HTTP endpoint available outside of Snowflake. Either endpoint can be used to call a service.

# mv: snowflake.ml.model.ModelVersion
mv.list_services()

{
    "data": [
                [0, 10, "Alex", "2014-01-01 16:00:00"],
                [1, 20, "Steve", "2015-01-01 16:00:00"],
                [2, 30, "Alice", "2016-01-01 16:00:00"],
                [3, 40, "Adrian", "2017-01-01 16:00:00"]
            ]
}

Latency considerations¶

You can use the following options to run inference while hosting a model in Snowpark Container Services. They have different trade-offs with regard to latency expectation and use cases.

HTTP endpoints

HTTP endpoints are best for online or real-time inference and offer the lowest latency of your options. Traffic doesn't go through a warehouse or global Snowflake services, and input data is directly sent to the server after authentication.

SQL

SQL commands are best for batch inference. Input data is sent to the server from a warehouse, which compiles and executes the query. Response data is then written to a cache in the warehouse before being returned to the server. Warehouse transfers, compilation, and execution incur latency. Execution latency is always lower when a cached result is used.

Snowpark Python APIs

Python APIs are backed by SQL commands executed on Snowflake and best-suited for the same purposes as direct SQL. However, latency is affected by the type of DataFrame used as input. Snowpark DataFrames use data already available in Snowflake and can operate entirely in the same manner as direct SQL. Pandas DataFrames upload their in-memory data to a temporary Snowflake table, which incurs transfer latency.

For the lowest latency when using Python APIs, prepare your pandas DataFrames as Snowpark DataFrames before hosting your model.

注釈

Performance evaluations you conduct locally using a pandas DataFrame won't match performance with the same pandas DataFrame that is running on Snowpark Container Services, due to the latency of temporary table creation. For an accurate performance evaluation of your model, use a Snowpark DataFrame as input.

サービスを一時停止します。¶

サービスは最初、非アクティブ状態が30分経過すると一時停止するように構成されています。サービスは、スケジューリングと起動の遅延に従い、新しいリクエストを受け取ると再開します。スケジューリングの遅延はコンピューティングプールの可用性に依存し、起動の遅延はモデルのサイズに依存します。

サービスを手動で一時停止したり再開したりするには、 ALTERSERVICE コマンドを使用してください。

ALTER SERVICE my_service [ SUSPEND | RESUME ];

自動一時停止の動作を手動で設定するには、ALTERSERVICE コマンドをAUTO_SUSPEND_SECS パラメーターと一緒に使用します。

ALTER SERVICE my_service SET AUTO_SUSPEND_SECS = <timeout_in_seconds>;

自動一時停止をオフにするには、AUTO_SUSPEND_SECSを 0に設定します。自動一時停止を元に戻すには、300秒以上のタイムアウト時間を指定します。

詳細については、ALTERSERVICE をご参照ください。

モデルの削除¶

モデルやモデルのバージョンは、 SQL インターフェイスや Python API を使って通常通り管理することができますが、サービスによって（実行中であれ中断中であれ）使われているモデルやモデルのバージョンはドロップ（削除）することができません。モデルまたはモデルのバージョンを削除するには、まずサービスを削除します。

CPU-powered推論のための XGBoost モデルの展開¶

次のコードは、SPCS で推論のためにXGBoostモデルをデプロイし、デプロイされたモデルを推論に使用する主な手順を示しています。この例のノートブック は使用可能です 。

from snowflake.ml.registry import registry
from snowflake.ml.utils.connection_params import SnowflakeLoginOptions
from snowflake.snowpark import Session

from xgboost import XGBRegressor

# your model training code here output of which is a trained xgb_model

# Open model registry
reg = registry.Registry(session=session, database_name='my_registry_db', schema_name='my_registry_schema')

# Log the model in Snowflake Model Registry
model_ref = reg.log_model(
    model_name="my_xgb_forecasting_model",
    version_name="v1",
    model=xgb_model,
    conda_dependencies=["scikit-learn","xgboost"],
    sample_input_data=pandas_test_df,
    comment="XGBoost model for forecasting customer demand"
)

# Deploy the model to SPCS
model_ref.create_service(
    service_name="forecast_model_service",
    service_compute_pool="my_cpu_pool",
    ingress_enabled=True)

# See all services running a model
model_ref.list_services()

# Run on SPCS
model_ref.run(pandas_test_df, function_name="predict", service_name="forecast_model_service")

# Delete the service
model_ref.delete_service("forecast_model_service")

Since this model has ingress enabled, you can call its public HTTP endpoint. The following example uses a PAT stored in the environment variable PAT_TOKEN to authenticate with a public Snowflake endpoint:

import os
import json
import numpy as np
from pprint import pprint
import requests

def get_headers(pat_token):
    headers = {'Authorization': f'Snowflake Token="{pat_token}"'}
    return headers

headers = get_headers(os.getenv("PAT_TOKEN"))

# Put the endpoint url with method name `predict`
# The endpoint url can be found with `show endpoints in service <service_name>`.
URL = 'https://<random_str>-<organization>-<account>.snowflakecomputing.app/predict'

# Prepare data to be sent
data = {"data": np.column_stack([range(pandas_test_df.shape[0]), pandas_test_df.values]).tolist()}

# Send over HTTP
def send_request(data: dict):
    output = requests.post(URL, json=data, headers=headers)
    assert (output.status_code == 200), f"Failed to get response from the service. Status code: {output.status_code}"
    return output.content

# Test
results = send_request(data=data)
print(json.loads(results))

GPU-powered推論のためのHugging Face文型変換器の展開¶

次のコードは、 HTTP エンドポイントを含むHugging Face文型変換器をトレーニングおよび展開します。

この例では、 sentence-transformers パッケージ、 GPU コンピューティングプール、イメージリポジトリが必要です。

from snowflake.ml.registry import registry
from snowflake.ml.utils.connection_params import SnowflakeLoginOptions
from snowflake.snowpark import Session
from sentence_transformers import SentenceTransformer

session = Session.builder.configs(SnowflakeLoginOptions("connection_name")).create()
reg = registry.Registry(session=session, database_name='my_registry_db', schema_name='my_registry_schema')

# Take an example sentence transformer from HF
embed_model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')

# Have some sample input data
input_data = [
    "This is the first sentence.",
    "Here's another sentence for testing.",
    "The quick brown fox jumps over the lazy dog.",
    "I love coding and programming.",
    "Machine learning is an exciting field.",
    "Python is a popular programming language.",
    "I enjoy working with data.",
    "Deep learning models are powerful.",
    "Natural language processing is fascinating.",
    "I want to improve my NLP skills.",
]

# Log the model with pip dependencies
pip_model = reg.log_model(
    embed_model,
    model_name="sentence_transformer_minilm",
    version_name="pip",
    sample_input_data=input_data,  # Needed for determining signature of the model
    pip_requirements=["sentence-transformers", "torch", "transformers"], # If you want to run this model in the Warehouse, you can use conda_dependencies instead
)

# Force Snowflake to not try to check warehouse
conda_forge_model = reg.log_model(
    embed_model,
    model_name="sentence_transformer_minilm",
    version_name="conda_forge_force",
    sample_input_data=input_data,
    # setting any package from conda-forge is sufficient to know that it can't be run in warehouse
    conda_dependencies=["sentence-transformers", "conda-forge::pytorch", "transformers"]
)

# Deploy the model to SPCS
pip_model.create_service(
    service_name="my_minilm_service",
    service_compute_pool="my_gpu_pool",  # Using GPU_NV_S - smallest GPU node that can run the model
    ingress_enabled=True,
    gpu_requests="1", # Model fits in GPU memory; only needed for GPU pool
    max_instances=4, # 4 instances were able to run 10M inferences from an XS warehouse
)

# See all services running a model
pip_model.list_services()

# Run on SPCS
pip_model.run(input_data, function_name="encode", service_name="my_minilm_service")

# Delete the service
pip_model.delete_service("my_minilm_service")

SQL では、次のようにサービス関数を呼び出すことができます：

SELECT my_minilm_service!encode('This is a test sentence.');

同様に、 HTTP エンドポイントを次のように呼び出すことができます。

import json
from pprint import pprint
import requests

# Put the endpoint url with method name `encode`
URL='https://<random_str>-<account>.snowflakecomputing.app/encode'

# Prepare data to be sent
data = {
    'data': []
}
for idx, x in enumerate(input_data):
    data['data'].append([idx, x])

# Send over HTTP
def send_request(data: dict):
    output = requests.post(URL, json=data, headers=headers)
    assert (output.status_code == 200), f"Failed to get response from the service. Status code: {output.status_code}"
    return output.content

# Test
results = send_request(data=data)
pprint(json.loads(results))

GPU-powered推論のための PyTorch モデルの展開¶

` ディープラーニング推奨モデル（<https://quickstarts.snowflake.com/guide/getting-started-with-running-distributed-pytorch-models-on-snowflake/>）を `_ 推論のために PyTorch にトレーニングしてデプロイする例については、 DLRM次のクイックスタート SPCSGPU をご参照ください。

SPCS 展開の監視¶

次の SQL クエリを使用して起動中のサービスを検査することで、デプロイメントを監視できます。

SHOW SERVICES IN COMPUTE POOL my_compute_pool;

2つのジョブが開始されます。

• MODEL_BUILD_xxxxx：名前の衝突を避けるため、名前の最後の文字はランダム化されます。このジョブはイメージを構築し、イメージが構築された後に終了します。イメージがすでに存在する場合は、ジョブはスキップされます。

  ログは、パッケージの依存関係の衝突などの問題をデバッグするのに便利です。このジョブのログを見るには、最後の文字が同じになるようにしたうえで以下の SQL を起動します。

  CALL SYSTEM$GET_SERVICE_LOGS('MODEL_BUILD_xxxxx', 0, 'model-build');
  

  Copy

• MYSERVICE: create_service へのコールで指定されたサービス名。このジョブは、 MODEL_BUILD のジョブが成功またはスキップされた場合に開始されます。このジョブのログを見るには、以下の SQL を起動します。

  CALL SYSTEM$GET_SERVICE_LOGS('MYSERVICE', 0, 'model-inference');
  

  Copy

  ビルドジョブまたはサービスが削除されたために SYSTEM$GET_SERVICE_LOG 経由でログが利用可能でない場合、 イベントテーブル （有効な場合）をチェックしてログを見ることができます。

  SELECT RESOURCE_ATTRIBUTES, VALUE
  FROM <EVENT_TABLE_NAME>
  WHERE true
      AND timestamp > dateadd(day, -1, current_timestamp())  -- choose appropriate timestamp range
      AND RESOURCE_ATTRIBUTES:"snow.database.name" = '<db of the service>'
      AND RESOURCE_ATTRIBUTES:"snow.schema.name" = '<schema of the service>'
      AND RESOURCE_ATTRIBUTES:"snow.service.name" = '<Job or Service name>'
  AND RESOURCE_ATTRIBUTES:"snow.service.container.instance" = '0'  -- choose all instances or one particular
  AND RESOURCE_ATTRIBUTES:"snow.service.container.name" != 'snowflake-ingress' --skip logs from internal sidecar
  ORDER BY timestamp ASC;
  

  Copy

パッケージの競合¶

サービスコンテナーにインストールされるパッケージは、モデル本体と推論サーバーの2つのシステムによって決定されます。モデルの依存関係との競合を最小限にするために、推論サーバーは以下のパッケージのみを必要とします。

• gunicorn<24.0.0

• starlette<1.0.0

• uvicorn-standard<1.0.0

モデルの依存関係が、上記と同様に、 pip または conda のどちらかによって解決可能であることを確認してください。

モデルに conda_dependencies と pip_requirements の両方が設定されている場合、これらはcondaを介して以下のようにインストールされます。

チャネル

• conda-forge

• nodefaults

依存関係

• all_conda_packages

• pip:

  • all_pip_packages

Snowflakeはコンテナーイメージをビルドする際に conda-forge からAnacaondaパッケージを取得します。これは、Snowflakeのcondaチャンネルはウェアハウスでのみ利用可能であり、 defaults チャンネルではユーザーがAnacondaの利用規約に同意する必要があるためですが、自動ビルド中には利用できません。defaults のような別のチャンネルからパッケージを取得するには、 defaults::pkg_name のように、各パッケージにチャンネル名を指定します。

メモリ不足のサービス¶

一部のモデルはスレッドセーフではないため、Snowflakeはワーカープロセスごとにモデルの別個のコピーをメモリにロードします。これは、ワーカー数の多い大規模なモデルでは、メモリ不足を引き起こす可能性があります。num_workers の削減を試します。

不満足のクエリパフォーマンス¶

通常、推論は推論サービスのインスタンス数がボトルネックになります。モデルを展開する際に、 max_instances に高い値を渡してみてください。

サービス仕様を変更できない¶

モデル構築および推論サービスの仕様は、 ALTER SERVICE を使用して変更することはできません。変更できるのは、 TAG 、 MIN_INSTANCES などの属性のみです。しかし、イメージはイメージ・レポで公開されているので、スペックをコピーして修正し、そこから新しいサービスを作成して手動で起動することができます。

パッケージが見つかりません¶

イメージ構築フェーズ中にモデルの展開に失敗しました。model-build ログによると、リクエストされたパッケージが見つかりませんでした。（このステップでは、パッケージが conda_dependencies に記載されている場合、デフォルトで conda-forge を使用します。）

パッケージのインストールは以下のような理由で失敗することがあります。

• パッケージ名またはバージョンが無効です。パッケージのスペルとバージョンを確認してください。

• リクエストされたバージョンのパッケージは conda-forge に存在しません。バージョン指定を削除して、 conda-forge で利用可能な最新バージョンを取得するか、代わりに pip_requirements を使用してみてください。こちら からすべての利用可能なパッケージを閲覧できます。

• 特別なチャンネル（例: pytorch）からのパッケージが必要な場合もあります。pytorch::torch のように、 channel_name:: のプレフィックスを依存関係に追加してください。

Huggingface Hubのバージョン不一致¶

Hugging Faceモデル推論サービスがエラーメッセージで失敗することがあります。

ImportError: huggingface-hub>=0.30.0,<1.0 is required for a normal functioning of this module, but found huggingface-hub==0.25.2

これは、 transformers パッケージが、 huggingface-hub への正しい依存関係を指定せず、代わりにコードでチェックするためです。この問題を解決するには、モデルを再度ログに記録し、今度は conda_dependencies または pip_requirements で huggingface-hub の必要なバージョンを明示的に指定します。

torchが CUDA を有効にしてコンパイルされていません¶

このエラーの典型的な原因は、 conda_dependencies と pip_requirements の両方を指定していることです。パッケージの競合 のセクションで述べたように、 conda はコンテナーイメージのビルドに使用するパッケージ管理です。Anacondaは conda_dependencies と pip_requirements のパッケージを一緒に解決せず、condaパッケージを優先します。これは、condaパッケージがpipパッケージと互換性がないという事態を引き起こしかねません。conda_dependencies ではなく pip_requirements で torch を指定している可能性があります。依存関係を conda_dependencies または pip_requirements に統合することを検討してください。それが不可能な場合は、 conda_dependencies で最も重要なパッケージを指定してください。