Gestion de Snowpark Container Services (y compris les fonctions de service) avec Python¶

Vous pouvez utiliser Python pour gérer Snowpark Container Services, un service de conteneurs entièrement géré par lequel vous pouvez déployer, gérer et mettre à l’échelle des applications conteneurisées. Pour une vue d’ensemble de Snowpark Container Services, voir À propos de Snowpark Container Services.

Avec les Snowflake Python APIs, vous pouvez gérer des pools de calcul, des référentiels d’images et des services.

Conditions préalables¶

Les exemples de cette rubrique supposent que vous ayez ajouté le code nécessaire pour vous connecter à Snowflake et créer un objet Root à partir duquel utiliser les Snowflake Python APIs.

Par exemple, le code suivant utilise les paramètres de connexion définis dans un fichier de configuration pour créer une connexion à Snowflake.

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

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

Copy

En utilisant l’objet Session obtenu, le code crée un objet Root pour utiliser les types et les méthodes de l’API. Pour plus d’informations, voir Connexion à Snowflake avec Snowflake Python APIs.

Gestion des pools de calcul¶

Vous pouvez gérer des pools de calcul, qui sont des collections de nœuds de machines virtuelles (VM) sur lesquels Snowflake exécute vos tâches et services Snowpark Container Services.

Les Snowflake Python APIs représentent des pools de calcul de deux types distincts :

ComputePool : affiche les propriétés d’un pool de calcul, telles que son entrepôt, le nombre de nœuds maximum et minimum, ainsi que les paramètres de reprise et de suspension automatiques.
ComputePoolResource : expose des méthodes permettant d’effectuer des actions sur les pools de calcul, telles que la récupération d’un objet ComputePool correspondant, ainsi que la suspension, la reprise et l’arrêt des pools.

Pour plus d’informations sur les pools de calcul, voir Snowpark Container Services : utilisation des pools de calcul.

Création d’un pool de calcul¶

Vous pouvez créer une base de données en appelant la méthode ComputePoolCollection.create et en lui transmettant un objet ComputePool représentant le pool de calcul que vous souhaitez créer.

Pour créer un pool de calcul, créez d’abord un objet ComputePool qui spécifie les propriétés du pool comme suit :

Nom du pool de calcul
Nombre maximum et minimum de nœuds que le pool contiendra
Nom de la famille d’instances identifiant le type de machine à provisionner pour les nœuds du pool
Si le pool doit reprendre automatiquement lorsqu’un service ou une tâche lui est soumis

Le code de l’exemple suivant crée un objet ComputePool représentant un pool nommé 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

Le code crée ensuite le pool de calcul en transmettant l’objet ComputePool à la méthode ComputePoolCollection.create.

Obtention des détails d’un pool de calcul¶

Vous pouvez obtenir des informations sur un pool de calcul en appelant la méthode ComputePoolResource.fetch qui renvoie un objet ComputePool.

Le code de l’exemple suivant permet d’obtenir des informations sur un pool de calcul nommé my_compute_pool :

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

Copy

Répertorier les pools de calcul¶

Vous pouvez répertorier les pools de calcul en utilisant la méthode iter, qui renvoie un itérateur PagedIter.

Le code de l’exemple suivant répertorie les pools de calcul dont le nom commence par my :

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

Copy

Effectuer des opérations sur le pool de calcul¶

Vous pouvez effectuer des opérations courantes sur les pools de calcul, telles que la suspension, la reprise et l’arrêt des pools, à l’aide d’un objet ComputePoolResource qui vous pouvez obtenir en utilisation la méthode ComputePool.fetch.

Le code de l’exemple suivant suspend, reprend et arrête le pool de calcul 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

Le code utilise la méthode Root.compute_pools pour créer un objet ComputePool représentant le pool de calcul. À partir de l’objet ComputePool , il récupère un objet ComputePoolResource avec lequel il effectue des opérations de pool de calcul.

Gestion des référentiels d’images¶

Vous pouvez gérer des référentiels d’images, qui stockent des images pour les applications que vous exécutez sur des services de conteneurs.

Un référentiel d’images est un objet de niveau schéma. Lorsque vous créez ou faites référence à un référentiel, vous le faites dans le contexte de son schéma.

Les Snowflake Python APIs représentent des référentiels d’images de deux types distincts :

ImageRepository : expose les propriétés d’un référentiel d’images telles que les noms de sa base de données et de son schéma, l’URL du référentiel et le propriétaire.
ImageRepositoryResource : expose des méthodes que vous pouvez utiliser pour extraire un objet ImageRepository correspondant et pour supprimer la ressource de référentiel d’images.

Pour plus d’informations sur les référentiels d’images, voir Snowpark Container Services : utilisation d’un registre et d’un référentiel d’images.

Création d’un référentiel d’images¶

Pour créer un référentiel d’images, il faut d’abord créer un objet ImageRepository qui spécifie le nom du référentiel.

Le code de l’exemple suivant crée un objet ImageRepository représentant un référentiel nommé 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

Le code crée ensuite le référentiel d’images en transmettant l’objet ImageRepository à la méthode ImageRepositoryCollection.create , créant ainsi le référentiel d’images dans la base de données my_db et le schéma my_schema.

Obtention des détails du référentiel d’images¶

Vous pouvez obtenir des informations sur un référentiel d’images en appelant la méthode ImageRepositoryResource.fetch , qui renvoie un objet ImageRepository.

Le code de l’exemple suivant récupère un objet ImageRepository représentant le référentiel d’images my_repo et affiche le nom du propriétaire du référentiel :

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

Répertorier des référentiels d’images¶

Vous pouvez répertorier les référentiels d’images d’un schéma spécifié via la méthode iter, qui renvoie un itérateur PagedIter d’objets ImageRepository.

Le code de l’exemple suivant répertorie les noms des référentiels dans la base de données my_db et le schéma 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

Suppression d’un référentiel d’images¶

Vous pouvez supprimer un référentiel d’images en utilisant la méthode ImageRepositoryResource.drop.

Le code dans l’exemple suivant supprime le référentiel my_repo :

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

Copy

Gestion des services et des fonctions de service¶

Vous pouvez gérer les services, qui exécutent les conteneurs d’application jusqu’à ce que vous les arrêtiez. Snowflake redémarre automatiquement un service si le conteneur de service s’arrête. De cette manière, le service fonctionne effectivement sans interruption.

Un service est un objet de niveau schéma. Lorsque vous créez ou faites référence à un service, vous le faites dans le contexte de son schéma.

Les Snowflake Python APIs représentent des services avec deux types distincts :

Service : expose les propriétés d’un service telles que sa spécification, le nombre d’instances minimum et maximum, le nom de la base de données et du schéma.
ServiceResource : expose des méthodes que vous pouvez utiliser pour récupérer un objet Service correspondant, suspendre et reprendre le service et obtenir son statut.

Pour plus d’informations sur les services, voir Snowpark Container Services : utilisation des tâches.

Création d’un service¶

Pour créer un service, vous exécutez la méthode services.create , en transmettant un objet Service représentant le service que vous souhaitez créer.

Vous créez un service à partir d’un fichier de spécification de service .yaml qui a été chargé dans une zone de préparation. Pour plus d’informations sur la création d’une spécification de service, voir Référence Spécification de service.

Chargement de la spécification¶

Si vous créez un service à partir d’une spécification qui n’a pas encore été chargée dans une zone de préparation, vous pouvez charger la spécification à l’aide d’un objet Snowpark FileOperation.

Le code de l’exemple suivant utilise la méthode FileOperation.put pour charger une spécification sous forme de fichier :

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

Copy

Le code de l’exemple suivant utilise la méthode FileOperation.put_stream pour charger une spécification sous forme de chaîne :

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

Copy

Création du service¶

Pour créer un service à partir d’une spécification en zone de préparation, créez d’abord un objet Service qui spécifie les propriétés du service comme suit :

Nom du service
Nombre maximum et minimum d’instances de service que Snowflake peut créer
Pool de calcul auquel le service doit être ajouté
Emplacement de la zone de préparation et nom de la spécification

Le code de l’exemple suivant crée un objet Service représentant un service nommé my_service à partir d’une spécification dans @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

Le code crée ensuite le service en transmettant l’objet Service à la méthode ServiceCollection.create , créant ainsi le service dans la base de données my_db et le schéma my_schema.

Vous pouvez également créer un service à partir d’une spécification que vous fournissez sous forme de texte en ligne, comme le montre l’exemple suivant. La fonction ServiceSpec prend un seul argument de chaîne spec. Si la chaîne commence par @, la fonction l’interprète et la valide comme un chemin de fichier de zone de préparation. Sinon, la chaîne est transmise en tant que texte en ligne.

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

Création d’une fonction de service¶

Une fois le service opérationnel, vous pouvez créer une fonction de service qui communique avec le point de terminaison de service. Une fonction de service est une fonction définie par l’utilisateur (UDF) que vous créez et associez à un service dans Snowpark Container Services. Pour plus d’informations, voir Fonctions de service : utilisation d’un service à partir d’une requête SQL.

Le code de l’exemple suivant crée une UDF nommée my-udf qui spécifie le service hello-world et le point de terminaison hello-world-endpoint que vous avez précédemment défini :

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

Appel d’une fonction de service¶

Une fois la fonction de service créée, vous pouvez alors appeler la fonction pour la tester.

Le code de l’exemple suivant appelle la fonction de service my-udf que vous avez créée précédemment :

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

Copy

Obtention d’informations sur les services¶

Vous pouvez obtenir des informations sur un service Snowflake en appelant la méthode ServiceResource.fetch , qui renvoie un objet Service.

Le code de l’exemple suivant permet d’obtenir des informations sur un service nommé my_service :

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

Copy

Répertorier des services¶

Vous pouvez répertorier les services d’un schéma spécifié en utilisant la méthode iter qui renvoie un itérateur PagedIter d’objets Service.

Le code de l’exemple suivant répertorie les services dont le nom commence par my :

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

Copy

Effectuer des opérations de service¶

Vous pouvez effectuer des opérations de service courantes, telles que la suspension, la reprise et l’obtention du statut du service, à l’aide d’un objet ServiceResource.

Le code de l’exemple suivant suspend et reprend le service my_service puis récupère le statut du 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)

Copy