Gerenciamento do Snowpark Container Services (incluindo funções de serviço) com Python¶

Você pode usar Python para gerenciar o Snowpark Container Services, um serviço de contêiner totalmente gerenciado por meio do qual você pode implantar, gerenciar e dimensionar aplicativos em contêineres. Para obter uma visão geral do Snowpark Container Services, consulte Sobre Snowpark Container Services.

Com o Snowflake Python APIs, é possível gerenciar pools de computação, repositórios de imagens e serviços.

Pré-requisitos¶

Os exemplos neste tópico pressupõem que você tenha adicionado código para se conectar ao Snowflake e criar um objeto Root a partir do qual usar o Snowflake Python APIs.

Por exemplo, o seguinte código usa parâmetros de conexão definidos em um arquivo de configuração para criar uma conexão com o Snowflake:

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)

Copy

Usando o objeto Session resultante, o código cria um objeto Root para usar os tipos e métodos de API. Para obter mais informações, consulte Conexão ao Snowflake com o Snowflake Python APIs.

Gerenciamento de pools de computação¶

Você pode gerenciar pools de computação, que são coleções de nós de máquinas virtuais (VM) nos quais o Snowflake executa seus trabalhos e serviços do Snowpark Container Services.

O Snowflake Python APIs representa pools de computação com dois tipos separados:

ComputePool: expõe as propriedades de um pool de computação, como seu warehouse, nós máximo e mínimo e configurações de retomada automática e suspensão automática.
ComputePoolResource: expõe métodos para executar ações em pools de computação, como buscar um objeto ComputePool correspondente e suspender, retomar e parar pools.

Para obter mais informações sobre pools de computação, consulte Snowpark Container Services: como trabalhar com pools de computação.

Criação de um pool de computação¶

Você pode criar um pool de computação chamando o método ComputePoolCollection.create, passando um objeto ComputePool que representa o pool de computação que você deseja criar.

Para criar um pool de computação, primeiro crie um objeto ComputePool que especifique propriedades do pool, como as seguintes:

Nome do pool de computação
Número máximo e mínimo de nós que o pool conterá
Nome da família de instâncias que identifica o tipo de máquina a ser provisionada para nós no pool
Se o pool deve ser retomado automaticamente quando um serviço ou trabalho é enviado a ele

O código no exemplo a seguir cria um objeto ComputePool que representa um pool nomeado my_compute_pool:

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)

Copy

O código então cria o banco de pool de computação passando o objeto ComputePool para o método ComputePoolCollection.create.

Como obter detalhes do pool de computação¶

Você pode obter informações sobre um pool de computação chamando o método ComputePoolResource.fetch, que retorna um objeto ComputePool.

O código no exemplo a seguir obtém informações sobre um pool chamado my_compute_pool:

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

Copy

Listagem de pools de computação¶

Você pode listar pools de computação usando o método iter, que retorna um iterador PagedIter.

O código no exemplo a seguir lista pools de computação cujos nomes começam com my:

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

Copy

Execução de operações de pool de computação¶

Você pode executar operações comuns de pool de computação,—como suspender, retomar e interromper pools—com um objeto ComputePoolResource, que você pode obter usando o método ComputePool.fetch.

O código no exemplo a seguir suspende, retoma e interrompe o pool de computação 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()

Copy

O código usa o método Root.compute_pools para criar um objeto ComputePool que representa o pool de computação. Do objeto ComputePool, ele busca um objeto ComputePoolResource com o qual executará operações de pool de computação.

Gerenciamento de repositórios de imagens¶

Você pode gerenciar repositórios de imagens, que armazenam imagens de aplicativos executados em serviços de contêiner.

Um repositório de imagens é um objeto no nível do esquema. Ao criar ou referenciar um repositório, você faz isso no contexto de seu esquema.

O Snowflake Python APIs representa repositórios de imagens com dois tipos separados:

ImageRepository: expõe as propriedades de um repositório de imagens, como banco de dados e nomes de esquema, URL do repositório e proprietário.
ImageRepositoryResource: Expõe métodos que podem ser usados para buscar um objeto ImageRepository correspondente e descartar o recurso do repositório de imagens.

Para obter mais informações sobre repositórios de imagens, consulte Snowpark Container Services: como trabalhar com um registro e repositório de imagens.

Criação de um repositório de imagens¶

Para criar um repositório de imagens, primeiro crie um objeto ImageRepository que especifique o nome do repositório.

O código no exemplo a seguir cria um objeto ImageRepository que representa um repositório nomeado my_repo:

from snowflake.core.image_repository import ImageRepository

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

Copy

O código então cria o repositório de imagens passando o objeto ImageRepository para o método ImageRepositoryCollection.create, criando o repositório de imagens no banco de dados my_db e no esquema my_schema.

Como obter detalhes do repositório de imagens¶

Você pode obter informações sobre um repositório de imagens chamando o método ImageRepositoryResource.fetch, que retorna um objeto ImageRepository.

O código no exemplo a seguir obtém um objeto ImageRepository que representa o repositório de imagens my_repo e então imprime o nome do proprietário do repositório:

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

Copy

Listagem do repositórios de imagens¶

Você pode listar os repositórios de imagens em um esquema especificado usando o método iter, que retorna um iterador PagedIter dos objetos ImageRepository.

O código no exemplo a seguir lista os nomes dos repositórios no banco de dados my_db e no esquema my_schema:

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

Copy

Descarte de um repositório de imagens¶

É possível descartar um repositório de imagens usando o método ImageRepositoryResource.drop.

O código no exemplo a seguir descarta o repositório my_repo:

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

Copy

Gerenciamento de serviços e funções de serviço¶

Você pode gerenciar serviços, que executam contêineres de aplicativos até serem interrompidos. O Snowflake reinicia um serviço automaticamente se o contêiner de serviço parar. Dessa forma, o serviço funciona efetivamente de forma ininterrupta.

Um serviço é um objeto no nível do esquema. Ao criar ou referenciar um serviço, você faz isso no contexto de seu esquema.

O Snowflake Python APIs representa serviços com dois tipos distintos:

Service: expõe as propriedades de um serviço, como especificação, instâncias mínimas e máximas e nome do banco de dados e do esquema.
ServiceResource: expõe métodos que você pode usar para buscar um objeto Service correspondente, suspender e retomar o serviço e excluir seu status.

Para obter mais informações sobre serviços, consulte Snowpark Container Services: como trabalhar com serviços.

Criação de um serviço¶

Para criar um serviço, você executa o método services.create, passando um objeto Service que representa o serviço que deseja criar.

Você cria um serviço a partir de um arquivo de especificação de serviço .yaml que foi transferido por upload para um estágio. Para obter mais informações sobre a criação de uma especificação do serviço, consulte Referência de especificação de serviço.

Carregamento da especificação¶

Se você estiver criando um serviço a partir de uma especificação que ainda não foi carregada em um estágio, poderá fazer upload da especificação usando um objeto FileOperation do Snowpark.

O código no exemplo a seguir usa o método FileOperation.put para fazer upload de uma especificação como um arquivo:

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

Copy

O código no exemplo a seguir usa o método FileOperation.put_stream para fazer upload de uma especificação como uma cadeia de caracteres:

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

Copy

Criação do serviço¶

Para criar um serviço a partir de uma especificação preparada, primeiro crie um objeto Service que especifique propriedades de serviço como estas:

Nome do serviço
Número máximo e mínimo de instâncias de serviço que o Snowflake pode criar
Pool de computação ao qual o serviço deve ser adicionado
Localização do estágio e nome da especificação

O código no exemplo a seguir cria um objeto Service que representa um serviço nomeado my_service a partir de uma especificação em @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)

Copy

O código então cria o serviço passando o objeto Service para o método ServiceCollection.create, criando o serviço no banco de dados my_db e no esquema my_schema.

Você também pode criar um serviço a partir de uma especificação fornecida como texto inline, conforme mostrado no exemplo a seguir. A função ServiceSpec aceita um único argumento da cadeia de caracteres spec. Se a sequência começar com @, a função interpreta e valida como um caminho de arquivo de estágio. Caso contrário, a cadeia de caracteres é passada como texto inline.

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)

Copy

Criação de uma função de serviço¶

Depois que o serviço estiver instalado e funcionando, você poderá criar uma função de serviço que se comunique com o ponto de extremidade do servidor. Uma função de serviço é uma função definida pelo usuário (UDF) que você cria e associa a um serviço no Snowpark Container Services. Para obter mais informações, consulte Funções de serviço: como usar um serviço de uma consulta SQL.

O código no exemplo a seguir cria uma UDF nomeada my-udf que especifica o serviço hello-world e o ponto de extremidade hello-world-endpoint que você definiu anteriormente:

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
)

Copy

Invocação de uma função de serviço¶

Depois que a função de serviço for criada, será possível invocar a função para testá-la.

O código no exemplo a seguir invoca a função de serviço my-udf que você criou anteriormente:

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

Copy

Como obter detalhes do serviço¶

Você pode obter informações sobre um serviço do Snowflake chamando o método ServiceResource.fetch, que retorna um objeto Service.

O código no exemplo a seguir obtém informações sobre um serviço nomeado my_service:

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

Copy

Listagem de serviços¶

Você pode listar os serviços em um esquema especificado usando o método iter, que retorna um iterador PagedIter dos objetos Service.

O código no exemplo a seguir lista serviços cujos nomes começam com my:

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

Copy

Execução de operações de serviço¶

Você pode realizar operações de serviço comuns,—como suspender, retomar e obter o status do serviço—com um objeto ServiceResource.

O código no exemplo a seguir suspende e retoma o serviço my_service e também obtém o status do serviço:

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)

Copy