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

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

Você pode definir as propriedades de um objeto ComputePool e passá-lo para o método ComputePoolResource.create_or_alter para criar um pool de computação, se ele não existir, ou alterá-lo de acordo com a definição do pool de computação, se ele existir. O comportamento do create_or_alter deve ser idempotente, o que significa que o objeto do pool de computação resultante será o mesmo, independentemente de o pool de computação existir ou não antes de você chamar o método.

Nota

O método create_or_alter usa valores padrão para todas as propriedades ComputePool que você não definir explicitamente. Por exemplo, se você não definir auto_resume, o valor padrão será None, mesmo que o pool de computação tenha existido anteriormente com um valor diferente.

O código no exemplo a seguir atualiza os nós máximos permitidos do pool de computação my_compute_pool e, em seguida, altera o pool de computação no 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)

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

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