Tâches Snowflake ML¶

Exécution d’une fonction Python en tant que tâche Snowflake ML¶

Utilisez Function Dispatch pour exécuter des fonctions Python individuelles à distance sur les ressources de calcul de Snowflake avec le décorateur @remote.

À l’aide du décorateur @remote, vous pouvez effectuer les opérations suivantes :

• Sérialiser la fonction et ses dépendances.

• La télécharger dans une zone de préparation Snowflake spécifiée.

• L’exécuter dans Container Runtime pour ML.

L’exemple de code Python suivant utilise le décorateur @remote pour soumettre une tâche Snowflake ML. Notez qu’une Session Snowpark est requise, voir Conditions préalables.

from snowflake.ml.jobs import remote

@remote("MY_COMPUTE_POOL", stage_name="payload_stage", session=session)
def train_model(data_table: str):
  # Provide your ML code here, including imports and function calls
  ...

job = train_model("my_training_data")

L’appel d’une fonction décorée @remote renvoie un objet MLJob Snowflake qui peut être utilisé pour gérer et surveiller l’exécution de la tâche. Pour plus d’informations, voir Gestion des tâches ML.

Exécuter un fichier Python en tant que tâche Snowflake ML¶

Exécutez des fichiers Python ou des répertoires de projets sur les ressources de calcul Snowflake. Ceci est utile lorsque :

• Vous avez des projets ML complexes avec de multiples modules et dépendances.

• Vous souhaitez maintenir une séparation entre le développement local et le code de production.

• Vous devez exécuter des scripts qui utilisent des arguments de ligne de commande.

• Vous travaillez avec des projets ML existants qui n’ont pas été spécifiquement conçus pour être exécutés sur le pool de calcul Snowflake.

L’API de la tâche Snowflake propose trois méthodes principales de soumission des charges utiles basées sur des fichiers :

• submit_file() : Pour l’exécution d’un seul fichier Python

• submit_directory() : Pour l’exécution de projets Python couvrant plusieurs fichiers et ressources

• submit_from_stage() : Pour l’exécution de projets Python enregistrés sur une zone de préparation Snowflake

Les deux méthodes prennent en charge ce qui suit :

• La transmission de l’argument de ligne de commande

• La configuration de la variable d’environnement

• La spécification de la dépendance personnalisée

• Gestion des actifs du projet à travers les zones de préparation Snowflake

File Dispatch est particulièrement utile pour mettre en production des workflows ML existants et pour maintenir une séparation claire entre les environnements de développement et d’exécution.

Le code Python suivant soumet un fichier en tant que tâche Snowflake ML :

from snowflake.ml.jobs import submit_file

# Run a single file
job1 = submit_file(
  "train.py",
  "MY_COMPUTE_POOL",
  stage_name="payload_stage",
  args=["--data-table", "my_training_data"],
  session=session,
)

Le code Python suivant soumet un répertoire en tant que tâche Snowflake ML :

from snowflake.ml.jobs import submit_directory

# Run from a directory
job2 = submit_directory(
  "./ml_project/",
  "MY_COMPUTE_POOL",
  entrypoint="train.py",
  stage_name="payload_stage",
  session=session,
)

Le code Python suivant soumet un répertoire à partir d’une zone de préparation Snowflake en tant que tâche Snowflake ML :

from snowflake.ml.jobs import submit_from_stage

# Run from a directory
job3 = submit_from_stage(
  "@source_stage/ml_project/"
  "MY_COMPUTE_POOL",
  entrypoint="@source_stage/ml_project/train.py",
  stage_name="payload_stage",
  session=session,
)

# Entrypoint may also be a relative path
job4 = submit_from_stage(
  "@source_stage/ml_project/",
  "MY_COMPUTE_POOL",
  entrypoint="train.py",  # Resolves to @source_stage/ml_project/train.py
  stage_name="payload_stage",
  session=session,
)

La soumission d’un fichier ou d’un répertoire renvoie un objet MLJob Snowflake qui peut être utilisé pour gérer et surveiller l’exécution de la tâche. Pour plus d’informations, voir Gestion des tâches ML.

Prise en charge des charges utiles supplémentaires dans les soumissions¶

Lors de la soumission d’un fichier, d’un répertoire ou à partir d’une zone de préparation, des charges utiles supplémentaires sont prises en charge pour une utilisation pendant l’exécution de la tâche. Le chemin d’importation peut être spécifié explicitement. Sinon, il sera déduit de l’emplacement de la charge utile supplémentaire.

# Run from a file
 job1 = submit_file(
   "train.py",
   "MY_COMPUTE_POOL",
   stage_name="payload_stage",
   session=session,
   additional_payloads=[
     ("src/utils/", "utils"), # the import path is utils
   ],
 )

 # Run from a directory
 job2 = submit_directory(
   "./ml_project/",
   "MY_COMPUTE_POOL",
   entrypoint="train.py",
   stage_name="payload_stage",
   session=session,
   additional_payloads=[
     ("src/utils/"), # the import path is utils
   ],
 )

 # Run from a stage
 job3 = submit_from_stage(
   "@source_stage/ml_project/",
   "MY_COMPUTE_POOL",
   entrypoint="@source_stage/ml_project/train.py",
   stage_name="payload_stage",
   session=session,
   additional_payloads=[
     ("@source_stage/src/utils/sub_utils/", "utils.sub_utils"),
   ],
 )

Accès à la session Snowpark dans les tâches ML¶

Lors de l’exécution des tâches ML sur Snowflake, une session Snowpark est automatiquement disponible dans le contexte d’exécution. Vous pouvez accéder à l’objet de la session à partir de la charge utile de la tâche ML à l’aide des approches suivantes :

from snowflake.ml.jobs import remote
from snowflake.snowpark import Session

@remote("MY_COMPUTE_POOL", stage_name="payload_stage")
def my_function():
  # This approach works for all payload types, including file and directory payloads
  session = Session.builder.getOrCreate()
  print(session.sql("SELECT CURRENT_VERSION()").collect())

@remote("MY_COMPUTE_POOL", stage_name="payload_stage")
def my_function_with_injected_session(session: Session):
  # This approach works only for function dispatch payloads
  # The session is injected automatically by the Snowflake ML Job API
  print(session.sql("SELECT CURRENT_VERSION()").collect())

La session Snowpark peut être utilisée pour accéder aux tableaux, zones de préparation et autres objets de base de données Snowflake à l’intérieur de votre tâche ML.

Renvoi des résultats des tâches ML¶

Les tâches Snowflake ML prennent en charge le renvoi des résultats de l’exécution vers l’environnement client. Cela vous permet de récupérer des valeurs calculées, des modèles entraînés ou tout autre artefact produit par vos charges utiles de tâche.

Pour la distribution des fonctions, il suffit de renvoyer une valeur de votre fonction décorée. La valeur renvoyée sera sérialisée et mise à disposition par l’intermédiaire de la méthode:code:result().

from snowflake.ml.jobs import remote

@remote("MY_COMPUTE_POOL", stage_name="payload_stage")
def train_model(data_table: str):
  # Your ML code here
  model = XGBClassifier()
  model.fit(data_table)
  return model

job1 = train_model("my_training_data")

Pour les tâches basées sur des fichiers, utilisez la variable spéciale __return__ pour spécifier la valeur de retour.

# Example: /path/to/repo/my_script.py
def main():
    # Your ML code here
    model = XGBClassifier()
    model.fit(data_table)
    return model

if __name__ == "__main__":
    __return__ = main()

from snowflake.ml.jobs import submit_file

job2 = submit_file(
    "/path/to/repo/my_script.py",
    "MY_COMPUTE_POOL",
    stage_name="payload_stage",
    session=session,
)

Vous pouvez récupérer le résultat de l’exécution de la tâche en utilisant l’API:code:MLJob.result(). L’API bloque le thread appelant jusqu’à ce que la tâche atteigne un état final, puis renvoie la valeur de retour de la charge utile ou, lève une exception en cas d’échec de l’exécution. Si la charge utile ne définit pas de valeur de retour, le résultat de la réussite sera None.

# These will block until the respective job is done and return the trained model
model1 = job1.result()
model2 = job2.result()

Flux de paquets privés¶

Les tâches Snowflake ML prennent également en charge le chargement de paquets à partir de flux privés tels que JFrog Artifactory et Sonatype Nexus Repository. Ces flux sont généralement utilisés pour distribuer des paquets internes et propriétaires, maintenir le contrôle sur les versions des dépendances et assurer la sécurité/conformité.

Pour installer des paquets à partir d’un flux privé, vous devez procéder comme suit :

1. Créez une règle de réseau pour autoriser l’accès à l’URL du flux privé.

   1. Pour les sources qui utilisent l’authentification de base, vous pouvez simplement créer une règle de réseau.

      CREATE OR REPLACE NETWORK RULE private_feed_nr
      MODE = EGRESS
      TYPE = HOST_PORT
      VALUE_LIST = ('<your-repo>.jfrog.io');
      

      Copy

   2. Pour configurer l’accès à une source à l’aide d’une connectivité privée (c’est-à-dire Private Link), suivez les étapes décrites dans Sortie réseau par connectivité privée.

2. Créez une intégration de l’accès externe à l’aide de la règle de réseau. Accordez l’autorisation d’utiliser l’EAI au rôle qui soumettra les tâches.

   CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION private_feed_eai
   ALLOWED_NETWORK_RULES = (PRIVATE_FEED_NR)
   ENABLED = true;
   
   GRANT USAGE ON INTEGRATION private_feed_eai TO ROLE <role_name>;
   

   Copy

3. Indiquez l’URL du flux privé, l’intégration de l’accès externe et le(s) paquet(s) lors de la soumission de la tâche

   # Option 1: Specify private feed URL in pip_requirements
   job = submit_file(
     "/path/to/script.py",
     compute_pool="MY_COMPUTE_POOL",
     stage_name="payload_stage",
     pip_requirements=[
       "--index-url=https://your.private.feed.url",
       "internal-package==1.2.3"
     ],
     external_access_integrations=["PRIVATE_FEED_EAI"]
   )
   

   Copy

   # Option 2: Specify private feed URL by environment variable
   job = submit_directory(
     "/path/to/code/",
     compute_pool="MY_COMPUTE_POOL",
     entrypoint="script.py",
     stage_name="payload_stage",
     pip_requirements=["internal-package==1.2.3"],
     external_access_integrations=["PRIVATE_FEED_EAI"],
     env_vars={'PIP_INDEX_URL': 'https://your.private.feed.url'},
   )
   

   Copy

Si votre flux privé URL contient des informations sensibles comme des jetons d’authentification, gérez l” URL en créant un secret Snowflake. Utilisez CREATE SECRET pour créer un secret. Configurez les secrets lors de la soumission d’une tâche avec l’argument spec_overrides.

# Create secret for private feed URL with embedded auth token
feed_url = "<your-repo>.jfrog.io/artifactory/api/pypi/test-pypi/simple"
user = "<auth_user>"
token = "<auth_token>"
session.sql(f"""
CREATE SECRET IF NOT EXISTS PRIVATE_FEED_URL_SECRET
 TYPE = GENERIC_STRING
 SECRET_STRING = 'https://{auth_user}:{auth_token}@{feed_url}'
""").collect()

# Prepare service spec override for mounting secret into job execution
spec_overrides = {
 "spec": {
  "containers": [
    {
     "name": "main",  # Primary container name is always "main"
     "secrets": [
      {
        "snowflakeSecret": "PRIVATE_FEED_URL_SECRET",
        "envVarName": "PIP_INDEX_URL",
        "secretKeyRef": "secret_string"
      },
     ],
    }
  ]
 }
}

# Load private feed URL from secret (e.g. if URL includes auth token)
job = submit_file(
  "/path/to/script.py",
  compute_pool="MY_COMPUTE_POOL",
  stage_name="payload_stage",
  pip_requirements=[
    "internal-package==1.2.3"
  ],
  external_access_integrations=["PRIVATE_FEED_EAI"],
  spec_overrides=spec_overrides,
)

Pour plus d’informations sur container.secrets, voir champ containers.secrets.