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())

コンピュートプールの作成と変更¶

ComputePool オブジェクトのプロパティをセットし、それを ComputePoolResource.create_or_alter メソッドに渡すことで、コンピュートプールが存在しない場合はそれを作成し、存在する場合はコンピュートプールの定義に従ってそれを変更することができます。 create_or_alter の動作は、べき等であることを意図しています。 つまり、メソッドを呼び出す前にコンピュートプールが存在していたかどうかに関係なく、結果として得られるコンピュートプール オブジェクトは同じになります。

以下の例のコードは、 my_compute_pool コンピュートプールの最大許容ノードを更新し、Snowflake上のコンピュートプールを変更します。

compute_pool = root.compute_pools["my_compute_pool"].fetch()
compute_pool.max_nodes = 3
compute_pool_res = root.compute_pools["my_compute_pool"].create_or_alter(compute_pool)

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

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()

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

イメージリポジトリを作成するには、まずリポジトリ名を指定する 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()
container_statuses = [container.status for container in my_service_res.get_containers()]