Python을 사용하여 Snowpark 컨테이너 서비스(서비스 함수 포함) 관리하기¶

컴퓨팅 풀 만들기¶

ComputePoolCollection.create 메서드를 호출하고 생성하려는 컴퓨팅 풀을 나타내는 ComputePool 오브젝트를 전달하여 컴퓨팅 풀을 생성할 수 있습니다.

컴퓨팅 풀을 생성하려면 먼저 다음과 같은 풀 속성을 지정하는 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)