PythonによるSnowpark Container Services（サービス関数を含む）の管理¶

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

作成するコンピューティングプールを表す ComputePool オブジェクトを渡して ComputePoolCollection.create メソッドを呼び出すと、コンピューティングプールを作成できます。

コンピューティングプールを作成するには、まず、以下のようなプールプロパティを指定する ComputePool オブジェクトを作成します。

• コンピュートプール名

• プールに含まれるノードの最大数と最小数

• プール内のノードにプロビジョニングするマシンのタイプを識別するインスタンスファミリーの名前

• サービスまたはジョブがプールに送信されたときに、プールを自動的に再開するかどうか。

次の例のコードでは、 my_compute_pool というプールを表す ComputePool オブジェクトを作成します。

from snowflake.core.compute_pool import ComputePool

compute_pool = ComputePool(name="my_compute_pool", min_nodes=1, max_nodes=2, instance_family="CPU_X64_XS", auto_resume=False)
root.compute_pools.create(compute_pool)

次に、コードは ComputePool オブジェクトを ComputePoolCollection.create メソッドに渡して、コンピューティングプールを作成します。

コンピュートプール詳細の取得¶

ComputePool オブジェクトを返す ComputePoolResource.fetch メソッドを呼び出すことで、コンピュートプールに関する情報を取得できます。

次の例のコードは、 my_compute_pool というプールに関する情報を取得します。

compute_pool = root.compute_pools["my_compute_pool"].fetch()
print(compute_pool.to_dict())

コンピュートプールの一覧表示¶

PagedIter 反復子を返す iter メソッドを使用して、コンピューティングプールをリストすることができます。

次の例のコードは、名前が my で始まるコンピューティングプールを一覧表示します。

compute_pools = root.compute_pools.iter(like="my%")
for compute_pool in compute_pools:
  print(compute_pool.name)

コンピュートプール操作の実行¶

ComputePool.fetch メソッドを使用して取得できる ComputePoolResource オブジェクトを使用して、プールの中断、再開、停止など一般的なコンピューティングプールの操作を実行できます。

次の例のコードは、 my_compute_pool コンピューティングプールを中断、再開、停止します。

compute_pool_res = root.compute_pools["my_compute_pool"]
compute_pool_res.suspend()
compute_pool_res.resume()
compute_pool_res.stop_all_services()

コードは、 Root.compute_pools メソッドを使用して、コンピューティングプールを表す ComputePool オブジェクトを作成します。 ComputePool オブジェクトから、 ComputePoolResource オブジェクトをフェッチし、コンピューティングプール操作を実行します。

イメージリポジトリの作成¶

イメージリポジトリを作成するには、まずリポジトリ名を指定する ImageRepository オブジェクトを作成します。

次の例のコードは、 my_repo という名前のリポジトリを表す ImageRepository オブジェクトを作成します。

from snowflake.core.image_repository import ImageRepository

my_repo = ImageRepository("my_repo")
root.databases["my_db"].schemas["my_schema"].image_repositories.create(my_repo)

次に、コードは ImageRepository オブジェクトを ImageRepositoryCollection.create メソッドに渡してイメージリポジトリを作成することで、 my_db データベースと my_schema スキーマにイメージリポジトリを作成します。

イメージリポジトリ詳細の取得¶

ImageRepository オブジェクトを返す ImageRepositoryResource.fetch メソッドを呼び出すことで、イメージリポジトリに関する情報を取得できます。

次の例のコードは、 my_repo イメージリポジトリを表す ImageRepository オブジェクトを取得して、リポジトリの所有者の名前を印刷します。

my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo = my_repo_res.fetch()
print(my_repo.owner)

イメージリポジトリの一覧表示¶

ImageRepository オブジェクトの PagedIter 反復子を返す iter メソッドを使用して、指定したスキーマのイメージリポジトリをリストすることができます。

次の例のコードは、 my_db データベースと my_schema スキーマのリポジトリ名をリストします。

repo_list = root.databases["my_db"].schemas["my_schema"].image_repositories.iter()
for repo_obj in repo_list:
  print(repo_obj.name)

イメージリポジトリのドロップ¶

ImageRepositoryResource.drop メソッドを使用してイメージリポジトリをドロップできます。

次の例のコードは、 my_repo リポジトリをドロップします。

my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo_res.drop()

サービスの作成¶

サービスを作成するには、 services.create メソッドを実行し、作成したいサービスを表す Service オブジェクトを渡します。

ステージにアップロードされたサービス仕様 .yaml ファイルからサービスを作成します。サービス仕様の作成に関する詳細については、 サービス仕様リファレンス をご参照ください。

session.file.put("/local_location/my_service_spec.yaml", "@my_stage")

service_spec_string = """
// Specification as a string.
"""
session.file.put_stream(StringIO(sepc_in_string), "@my_stage/my_service_spec.yaml")

from snowflake.core.service import Service, ServiceSpec

my_service = Service(name="my_service", min_instances=1, max_instances=2, compute_pool="my_compute_pool", spec=ServiceSpec("@my_stage/my_service_spec.yaml"))
root.databases["my_db"].schemas["my_schema"].services.create(my_service)

from textwrap import dedent
from snowflake.core.service import Service, ServiceSpec

spec_text = dedent(f"""\
    spec:
      containers:
      - name: hello-world
        image: repo/hello-world:latest
      endpoints:
      - name: hello-world-endpoint
        port: 8080
        public: true
    """)

my_service = Service(name="my_service", min_instances=1, max_instances=2, compute_pool="my_compute_pool", spec=ServiceSpec(spec_text))
root.databases["my_db"].schemas["my_schema"].services.create(my_service)

from snowflake.core import CreateMode
from snowflake.core.function import FunctionArgument, ServiceFunction

root.databases["my_db"].schemas["my_schema"].functions.create(
  ServiceFunction(
    name="my-udf",
    arguments=[
        FunctionArgument(name="input", datatype="TEXT")
    ],
    returns="TEXT",
    service="hello-world",
    endpoint="'hello-world-endpoint'",
    path="/hello-world-path",
    max_batch_rows=5,
  ),
  mode = CreateMode.or_replace
)

result = root.databases["my_db"].schemas["my_schema"].functions["my-udf(TEXT)"].execute_function(["test"])
print(result)

サービス詳細の取得¶

Service オブジェクトを返す ServiceResource.fetch メソッドを呼び出すことで、Snowflakeサービスに関する情報を取得できます。

次の例のコードは、 my_service というサービスに関する情報を取得します。

my_service = root.databases["my_db"].schemas["my_schema"].services["my_service"].fetch()

サービスの一覧表示¶

Service オブジェクトの PagedIter 反復子を返す iter メソッドを使用して、指定したスキーマのサービスをリストすることができます。

次の例のコードは、名前が my で始まるサービスを一覧表示します。

services = root.databases["my_db"].schemas["my_schema"].services.iter(like="my%")
for service_obj in services:
  print(service_obj.name)

サービス操作の実行¶

ServiceResource オブジェクトを使用して、サービスの中断、再開、そのステータスの取得など一般的なサービスの操作を実行できます。

次の例のコードは、 my_service サービスを中断および再開し、サービスのステータスも所得します。

my_service_res = root.databases["my_db"].schemas["my_schema"].services["my_service"]

my_service_res.suspend()
my_service_res.resume()
status = my_service_res.get_service_status(10)