Tutoriel 2 : créer une tâche Snowpark Container Services¶

Introduction¶

Après avoir terminé le tutoriel Configuration commune, vous êtes prêt à créer une tâche. Dans ce tutoriel, vous créez une tâche simple qui se connecte à Snowflake, exécute une requête SQL SELECT et enregistre le résultat dans une table.

Ce tutoriel comporte deux parties :

Partie 1 : Créer et tester une tâche. Vous téléchargez le code fourni pour ce tutoriel et suivez les instructions étape par étape :

1. Téléchargez le code de la tâche pour ce tutoriel.

2. Créez une image Docker pour Snowpark Container Services, et chargez l’image dans un référentiel de votre compte.

3. Mettez en zone de préparation le fichier de spécification de tâche, qui donne à Snowflake les informations de configuration du conteneur. En plus du nom de l’image à utiliser pour démarrer un conteneur, le fichier de spécification fournit :

   • Trois arguments : une requête SELECT, un entrepôt virtuel pour exécuter la requête et le nom de la table dans laquelle enregistrer le résultat.

   • L’entrepôt dans lequel l’instruction SELECT doit être exécutée.

4. Exécutez la tâche. En utilisant la commande EXECUTE SERVICE, vous pouvez exécuter la tâche en fournissant le fichier de spécification et le pool de calcul où Snowflake peut exécuter le conteneur. Enfin, vérifiez les résultats de la tâche.

Partie 2 : comprendre le code de la tâche. Cette section donne une vue d’ensemble du code de la tâche et met en évidence la manière dont les différents éléments collaborent.

1 : Télécharger le code de service¶

Un code (une application Python) est fourni pour créer une tâche.

1. Téléchargez le fichier zip dans un répertoire.

2. Décompressez le contenu, qui comprend un répertoire pour chaque tutoriel. Le répertoire Tutorial-2 contient les fichiers suivants :

   • main.py

   • Dockerfile

   • my_job_spec.yaml

2 : Construire et charger une image¶

Construisez une image pour la plateforme linux/amd64 prise en charge par Snowpark Container Services, puis chargez l’image dans le référentiel d’images de votre compte (voir Configuration commune).

Vous aurez besoin d’informations sur le référentiel (l’URL du référentiel et le nom d’hôte du registre) avant de pouvoir construire et charger l’image. Pour plus d’informations, voir Registre et référentiels.

Obtenir des informations sur le référentiel

1. Pour obtenir l’URL du référentiel, exécutez la commande SQL SHOW IMAGE REPOSITORIES.

   SHOW IMAGE REPOSITORIES;
   

   Copy

   • La colonne repository_url de la sortie fournit l’URL. En voici un exemple :

     <orgname>-<acctname>.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository
     

   • Le nom d’hôte dans l’URL du référentiel est le nom d’hôte du registre. En voici un exemple :

     <orgname>-<acctname>.registry.snowflakecomputing.com
     

Construire l’image et la charger dans le référentiel

1. Ouvrez une fenêtre de terminal et accédez au répertoire contenant les fichiers que vous avez décompressés.

2. Pour créer une image Docker, exécutez la commande docker build suivante à l’aide de la CLI Docker. Notez que la commande spécifie le répertoire de travail actuel (.) comme PATH pour les fichiers à utiliser pour la construction de l’image.

   docker build --rm --platform linux/amd64 -t <repository_url>/<image_name> .
   

   Copy

   • Pour image_name, utilisez my_job_image:latest.

   Exemple

   docker build --rm --platform linux/amd64 -t myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/my_job_image:latest .
   

   Copy

3. Chargez l’image dans le référentiel de votre compte Snowflake. Pour que Docker puisse charger une image en votre nom dans votre référentiel, vous devez d’abord authentifier Docker avec Snowflake.

   1. Pour authentifier Docker auprès du registre Snowflake, exécutez la commande suivante.

      docker login <registry_hostname> -u <username>
      

      Copy

      • Pour username, indiquez votre nom d’utilisateur Snowflake. Docker vous demandera votre mot de passe.

   2. Pour charger l’image, exécutez la commande suivante :

      docker push <repository_url>/<image_name>
      

      Copy

      Exemple

      docker push myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/my_job_image:latest
      

      Copy

3 : Mettre en zone de préparation le fichier de spécification¶

• Pour charger votre fichier de spécification de tâche (my_job_spec.yaml) sur la zone de préparation, utilisez l’une des options suivantes :

  • L’interface Web Snowsight : Pour les instructions, voir Choix d’une zone de préparation interne pour les fichiers locaux.

  • La CLI SnowSQL : exécutez la commande PUT suivante :

    PUT file://<file-path>[/\]my_job_spec.yaml @tutorial_stage
      AUTO_COMPRESS=FALSE
      OVERWRITE=TRUE;
    

    Copy

    Par exemple :

    • Linux ou macOS

      PUT file:///tmp/my_job_spec.yaml @tutorial_stage
        AUTO_COMPRESS=FALSE
        OVERWRITE=TRUE;
      

      Copy

    • Windows

      PUT file://C:\temp\my_job_spec.yaml @tutorial_stage
        AUTO_COMPRESS=FALSE
        OVERWRITE=TRUE;
      

      Copy

    Vous pouvez également spécifier un chemin relatif.

    PUT file://./my_job_spec.yaml @tutorial_stage
      AUTO_COMPRESS=FALSE
      OVERWRITE=TRUE;
    

    Copy

    La commande définit OVERWRITE=TRUE afin que vous puissiez charger le fichier à nouveau, si nécessaire (par exemple, si vous avez corrigé une erreur dans votre fichier de spécification). Si la commande PUT est exécutée avec succès, les informations relatives au fichier chargé sont affichées.

4 : Exécuter la tâche¶

Vous êtes maintenant prêt à créer une tâche.

1. Pour démarrer une tâche, exécutez la commande EXECUTE SERVICE :

   EXECUTE SERVICE
     IN COMPUTE POOL tutorial_compute_pool
     FROM @tutorial_stage
     SPEC='my_job_spec.yaml';
   

   Copy

   Remarques :

   • FROM et SPEC indiquent le nom de la zone de préparation et le nom du fichier de spécification de la tâche. Lorsque la tâche est exécutée, elle exécute l’instruction SQL et enregistre le résultat dans une table comme spécifié dans my_job_spec.yaml.

     L’instruction SQL de votre tâche n’est pas exécutée dans le conteneur Docker. Au lieu de cela, le conteneur en cours d’exécution se connecte à Snowflake et exécute l’instruction SQL dans un entrepôt Snowflake.

   • COMPUTE_POOL fournit les ressources informatiques où Snowflake exécute la tâche.

   • EXECUTE SERVICE renvoie une sortie qui inclut le UUID de la tâche attribué par Snowflake, comme le montre l’exemple de sortie suivant :

     +------------------------------------------------------------------------------------+
     |                      status                                                        |
     -------------------------------------------------------------------------------------+
     | Job 01af7ee6-0001-cb52-0020-c5870077223a completed successfully with status: DONE. |
     +------------------------------------------------------------------------------------+
     

2. Obtenez l’ID de la requête que vous avez exécutée (EXECUTE SERVICE est une requête).

   SET jobid = LAST_QUERY_ID();
   

   Copy

   Vous utilisez l’ID dans les étapes suivantes pour récupérer le statut du travail et les informations du journal des tâches.

   Note

   Il est important d’appeler LAST_QUERY_ID immédiatement après avoir appelé EXECUTE SERVICE pour s’assurer que l’ID de la tâche renvoyé par la commande est destiné à la commande EXECUTE SERVICE.

   LAST_QUERY_ID ne renvoie l’ID de requête d’une tâche qu’une fois la tâche terminée ; pour les tâches en cours de longue durée, cela ne convient pas pour obtenir des informations sur le statut en temps réel, et vous devriez plutôt utiliser la famille QUERY HISTORY de fonctions de table pour obtenir l’ID de requête de la tâche. Pour plus d’informations, voir Utilisation des tâches.

3. Cette tâche exécute une requête simple et enregistre les résultats dans la table des résultats. Vous pouvez vérifier que la tâche s’est bien déroulée en interrogeant la table des résultats :

   SELECT * FROM results;
   

   Copy

   Exemple de sortie :

   +----------+-----------+
   | TIME     | TEXT      |
   |----------+-----------|
   | 10:56:52 | hello     |
   +----------+-----------+
   

4. Si vous souhaitez déboguer l’exécution de votre tâche, utilisez les fonctions du système. Par exemple, utilisez SYSTEM$GET_JOB_STATUS pour déterminer si la tâche est toujours en cours d’exécution, si elle n’a pas démarré ou, le cas échéant, pourquoi elle a échoué. De plus, si votre code produit des journaux utiles sur la sortie standard ou l’erreur standard, vous pouvez accéder aux journaux en utilisant SYSTEM$GET_JOB_LOGS.

   1. Pour obtenir le statut d’une tâche, appelez la fonction système SYSTEM$GET_JOB_STATUS :

      SELECT SYSTEM$GET_JOB_STATUS($jobid);
      

      Copy

      Exemple de sortie :

      [
        {
            "status":"DONE",
            "message":"Completed successfully",
            "containerName":"main",
            "instanceId":"0",
            "serviceName":"01af7ee6-0001-cb52-0020-c5870077223a",
            "image":"orgname-acctname.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/my_job_image:latest",
            "restartCount":0,
            "startTime":""
        }
      ]
      

      Copy

      Dans la sortie, comme la tâche n’a pas de nom, le serviceName est le UUID attribué par Snowflake (ID de requête) de la tâche.

   2. Pour obtenir les informations du journal des tâches, utilisez la fonction système SYSTEM$GET_JOB_LOGS :

      SELECT SYSTEM$GET_JOB_LOGS($jobid, 'main')
      

      Copy

      job-tutorial - INFO - Job started
      job-tutorial - INFO - Connection succeeded. Current session context: database="TUTORIAL_DB", schema="DATA_SCHEMA", warehouse="TUTORIAL_WAREHOUSE", role="TEST_ROLE"
      job-tutorial - INFO - Executing query [select current_time() as time,'hello'] and writing result to table [results]
      job-tutorial - INFO - Job finished
      

5 : Nettoyage¶

Si vous ne prévoyez pas de poursuivre le tutoriel 3, vous devez supprimer les ressources facturables que vous avez créées. Pour plus d’informations, voir l’étape 5 dans le tutoriel 3.

6 : Vérification du code de tâche¶

Cette section couvre les sujets suivants :

• Vérification des fichiers fournis : examinez les différents fichiers de code qui mettent en œuvre la tâche.

• Construire et tester une image localement. Cette section explique comment vous pouvez tester localement l’image Docker avant de la charger vers un référentiel dans votre compte Snowflake.

#!/opt/conda/bin/python3

import argparse
import logging
import os
import sys

from snowflake.snowpark import Session
from snowflake.snowpark.exceptions import *

# Environment variables below will be automatically populated by Snowflake.
SNOWFLAKE_ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT")
SNOWFLAKE_HOST = os.getenv("SNOWFLAKE_HOST")
SNOWFLAKE_DATABASE = os.getenv("SNOWFLAKE_DATABASE")
SNOWFLAKE_SCHEMA = os.getenv("SNOWFLAKE_SCHEMA")

# Custom environment variables
SNOWFLAKE_USER = os.getenv("SNOWFLAKE_USER")
SNOWFLAKE_PASSWORD = os.getenv("SNOWFLAKE_PASSWORD")
SNOWFLAKE_ROLE = os.getenv("SNOWFLAKE_ROLE")
SNOWFLAKE_WAREHOUSE = os.getenv("SNOWFLAKE_WAREHOUSE")


def get_arg_parser():
  """
  Input argument list.
  """
  parser = argparse.ArgumentParser()
  parser.add_argument("--query", required=True, help="query text to execute")
  parser.add_argument(
    "--result_table",
    required=True,
    help="name of the table to store result of query specified by flag --query")

  return parser


def get_logger():
  """
  Get a logger for local logging.
  """
  logger = logging.getLogger("job-tutorial")
  logger.setLevel(logging.DEBUG)
  handler = logging.StreamHandler(sys.stdout)
  handler.setLevel(logging.DEBUG)
  formatter = logging.Formatter("%(name)s - %(levelname)s - %(message)s")
  handler.setFormatter(formatter)
  logger.addHandler(handler)
  return logger


def get_login_token():
  """
  Read the login token supplied automatically by Snowflake. These tokens
  are short lived and should always be read right before creating any new connection.
  """
  with open("/snowflake/session/token", "r") as f:
    return f.read()


def get_connection_params():
  """
  Construct Snowflake connection params from environment variables.
  """
  if os.path.exists("/snowflake/session/token"):
    return {
      "account": SNOWFLAKE_ACCOUNT,
      "host": SNOWFLAKE_HOST,
      "authenticator": "oauth",
      "token": get_login_token(),
      "warehouse": SNOWFLAKE_WAREHOUSE,
      "database": SNOWFLAKE_DATABASE,
      "schema": SNOWFLAKE_SCHEMA
    }
  else:
    return {
      "account": SNOWFLAKE_ACCOUNT,
      "host": SNOWFLAKE_HOST,
      "user": SNOWFLAKE_USER,
      "password": SNOWFLAKE_PASSWORD,
      "role": SNOWFLAKE_ROLE,
      "warehouse": SNOWFLAKE_WAREHOUSE,
      "database": SNOWFLAKE_DATABASE,
      "schema": SNOWFLAKE_SCHEMA
    }

def run_job():
  """
  Main body of this job.
  """
  logger = get_logger()
  logger.info("Job started")

  # Parse input arguments
  args = get_arg_parser().parse_args()
  query = args.query
  result_table = args.result_table

  # Start a Snowflake session, run the query and write results to specified table
  with Session.builder.configs(get_connection_params()).create() as session:
    # Print out current session context information.
    database = session.get_current_database()
    schema = session.get_current_schema()
    warehouse = session.get_current_warehouse()
    role = session.get_current_role()
    logger.info(
      f"Connection succeeded. Current session context: database={database}, schema={schema}, warehouse={warehouse}, role={role}"
    )

    # Execute query and persist results in a table.
    logger.info(
      f"Executing query [{query}] and writing result to table [{result_table}]"
    )
    res = session.sql(query)
    # If the table already exists, the query result must match the table scheme.
    # If the table does not exist, this will create a new table.
    res.write.mode("append").save_as_table(result_table)

  logger.info("Job finished")


if __name__ == "__main__":
  run_job()

ARG BASE_IMAGE=continuumio/miniconda3:4.12.0
FROM $BASE_IMAGE
RUN conda install python=3.8 && \
  conda install snowflake-snowpark-python
COPY main.py ./
ENTRYPOINT ["python3", "main.py"]

spec:
container:
- name: main
   image: /tutorial_db/data_schema/tutorial_repository/my_job_image:latest
   env:
      SNOWFLAKE_WAREHOUSE: tutorial_warehouse
   args:
   - "--query=select current_time() as time,'hello'"
   - "--result_table=results"

Quelle est la prochaine étape ?¶

Vous pouvez maintenant tester le tutoriel 3, qui montre comment fonctionne la communication de service à service.