Charger et utiliser des fonctions personnalisées dans les salles blanches de données de collaboration¶

Introduction¶

Tout collaborateur peut charger des UDFs et des UDTFs Python personnalisées vers une collaboration. Les modèles de la collaboration peuvent exécuter ces fonctions pour effectuer des actions de données complexes. L’utilisation courante inclut le machine learning ou la manipulation de données personnalisée dans une requête. Votre code chargé peut importer et utiliser des paquets à partir d’un bundle de packages Python approuvés et des paquets Snowpark API.

Les sections suivantes vous montrent comment charger et utiliser une fonction personnalisée.

Définir et utiliser des bundles de codes personnalisés¶

Voici comment charger et utiliser une fonction personnalisée :

Le demandeur de code :

1. Crée et enregistre le code en appelant REGISTER_CODE_SPEC.

   Le code peut être en ligne dans la spécification, ou lié à une zone de préparation.

2. Crée un modèle qui fait référence à la spécification du bundle de code par son ID dans le tableau code_specs du modèle. Ajoutez ce champ en tant que pair des champs du modèle et de paramètres, comme indiqué dans cet exemple :

   CopyExpand

    parameters:
      - name: <parameter_name>
        description: <parameter_description>
        required: <true_or_false>
        default: <default_value>
        type: <data_type>
   
    code_specs:             # Optional: List of code bundles used by this template
    - <code_spec_id>        # One or more code spec IDs.
   
    template: |
      <template_content>
   

   Show lessSee more

   Scroll to top

3. Enregistre le modèle, puis lie le modèle à la collaboration.

   Note

   Snowflake analyse le code importé pour détecter les problèmes de sécurité. Si un problème de sécurité est détecté, le code, ainsi que le modèle le contenant, ne seront pas ajoutés à la collaboration.

L’exécuteur d’analyses :

• Exécute le modèle de la manière standard en appelant RUN.

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_CODE_SPEC(
  $$
  api_version: 2.0.0
  spec_type: code_spec
  name: custom_udf
  version: v1
  functions:
    - name: normalize_value
      type: UDF
      language: PYTHON
      handler: normalize
      arguments:
        - name: value
          type: FLOAT
      returns: FLOAT
      code_body: |
        def normalize(value):
            return value / 100.0
  $$
);

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_TEMPLATE(
  $$
  api_version: 2.0.0
  spec_type: template
  name: normalization_template
  version: v1
  type: sql_analysis
  code_specs:
    - custom_udf_v1  -- Imports the code bundle.
  template: |
    SELECT cleanroom.custom_udf$normalize_value(metric, 0, 100)  -- Calls the UDF.
      AS normalized
        FROM {{ source_tables[0] }}
  $$
);

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.COLLABORATION.ADD_TEMPLATE_REQUEST(
  'my_collaboration',
  'normalization_template_v1',
  ['consumer']
);

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_TEMPLATE(
  $$
  api_version: 2.0.0
  spec_type: template
  name: normalization_template
  version: v1
  type: sql_analysis
  code_specs:
    - custom_udf_v1  -- Bundle ID includes the version number.
  template: |
    SELECT cleanroom.custom_udf$normalize_value(metric, 0, 100)  -- Calls the UDF.
      AS normalized
        FROM {{ source_tables[0] }}
  $$
);

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_TEMPLATE(
  $$
  api_version: 2.0.0
  spec_type: template
  name: normalization_template
  version: v2        -- Update the template version.
  type: sql_analysis
  code_specs:
    - custom_udf_v2  -- Use the new code bundle.
  template: |
    SELECT cleanroom.custom_udf$normalize_value(metric, 0, 100)  -- No change needed here.
      AS normalized
        FROM {{ source_tables[0] }}
  $$
);

Spécification du bundle de code¶

Cette spécification définit un bundle d’une ou plusieurs fonctions ou procédures de code qui peuvent être appelées par un modèle.

Une spécification de bundle de code comporte un maximum de 5 fonctions plus procédures.

Les identificateurs de la spécification du bundle de code présentent les exigences générales suivantes :

• Noms : Doivent être des identificateurs Snowflake valides qui commencent par une lettre et ne contiennent que des caractères alphanumériques et des traits de soulignement.

• Identificateurs avec guillemets : Les identificateurs entre guillemets doubles sont pris en charge pour les noms contenant des caractères spéciaux.

• Sensibilité à la casse : Les identificateurs sans guillemets sont insensibles à la casse ; les identificateurs entre guillemets préservent la casse.

api_version: 2.0.0              # Required: Must be "2.0.0"
spec_type: code_spec            # Required: Must be "code_spec"
name: <identifier>              # Required: Unique name of this code bundle.
version: <version_id>           # Required: Alphanumeric with underscores (max 20 chars)
description: <description_text> # Optional: Description (max 1,000 chars)

artifacts:                      # Optional: Staged files for import
  - alias: <identifier>         # One or more artifact items...
    stage_path: <stage_path>    # Required: Full stage path. See below for additional requirements.
    description: <description_text>  # Optional: Description (max 500 chars)
    content_hash: <sha256_hash>      # Optional: SHA-256 hash for integrity verification

functions:                      # Required if no procedures defined
  - name: <identifier>          # One or more functions...
    type: UDF | UDTF            # Required: Function type
    language: PYTHON            # Required: Currently only PYTHON supported
    runtime_version: <python_version>  # Optional: Python runtime (3.10 - 3.14)
    handler: <handler>          # Required: Handler function
    arguments:                  # Optional: One or more function arguments
      - name: <arg_name>        # Argument name
        type: <sql_type>        # Snowflake SQL type of this argument
    returns: <sql_type>         # Required: Snowflake return type
    packages:                   # Optional: Package dependencies
      - <package_name>          # One or more package items...
    imports:                    # Optional: Artifact aliases to import
      - <artifact_alias>        # One or more import items...
    code_body: |                # Optional: Inline Python code (max 12 MB)
      <inline_python_code>
    description: <description_text>  # Optional: Description of this function.

procedures:                     # Required if no functions defined
  - name: <identifier>          # One or more procedure items...
    language: PYTHON            # Required: Currently only PYTHON supported
    runtime_version: <python_version>  # Optional: Python runtime version
    handler: <handler>          # Required: Handler function
    arguments:                  # Optional: One or more procedure arguments
      - name: <arg_name>        # Argument name
        type: <sql_type>        # Snowflake SQL type of this argument
    returns: <sql_type>         # Optional: Return type
    packages:                   # Optional: Package dependencies
      - <package_name>          # One or more package items...
    imports:                    # Optional: Artifact aliases to import
      - <artifact_alias>        # One or more import items...
    code_body: |                # Optional: Inline Python code
      # inline python_code ...
    description: <description_text>  # Optional: Description of this procedure.

api_version

Version de l’API de collaboration utilisée. Doit être 2.0.0.

spec_type

Identificateur du type de spécification. Doit être code_spec.

name: identifier

Un nom unique pour cette spécification de bundle de code dans ce registre. Doit être un identificateur Snowflake valide avec un maximum de 75 caractères. Il est utilisé comme segment de nom de famille lors de l’appel de la fonction dans un modèle : cleanroom.code_spec_name$function_name

version: version_id

Identificateur de version personnalisée. Doit être alphanumérique avec des traits de soulignement, 20 caractères au maximum.

description: description_text (Facultatif)

Une description de la spécification du bundle de code (1 000 caractères maximum).

artifacts (Facultatif)

Une liste des fichiers ou des paquets en zone de préparation qui peuvent être importés par vos fonctions ou procédures, et :ref:` éventuellement exposés via des fonctions de gestionnaire <label-dcr_collab_code_bundles_staged_artifacts>`. Maximum de 5 par spécification.

alias: identifier

Un alias pour référencer cet artefact dans les importations. Lorsque vous faites référence à cet alias dans cette spécification, utilisez le nom d’alias brut plutôt que cleanroom.spec_name$alias ; autrement dit, utilisez le nom de la fonction vide pour référencer une autre fonction dans cette spécification.

stage_path: stage_path

Chemin complet de la zone de préparation vers le fichier d’artefact ; par exemple, @DB.SCHEMA.STAGE/path/file.whl.

• La zone de préparation doit être interne. Les zones de préparation externes ne sont pas prises en charge.

• La zone de préparation doit avoir DIRECTORY activé : La zone de préparation contenant les artefacts doit avoir DIRECTORY = TRUE défini.

• Format du chemin de la zone de préparation : Doit suivre le format @[DB.]SCHEMA.STAGE/path/to/file.ext.

• Pas de déplacement de chemin : Les chemins de zone de préparation ne peuvent pas contenir .. ou \.

• Cet artefact doit exister : Le fichier doit exister au niveau du chemin de la zone de préparation spécifié lors de l’enregistrement du bundle de code.

• La zone de préparation doit avoir le chiffrement coté serveur SNOWFLAKE_SSE activé. Lors de la création ou de la modification de la zone de préparation, définissez ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE').

• Si vous poussez, supprimez ou mettez à jour un fichier de code en zone de préparation, vous devez appeler ALTER STAGE stage name REFRESH pour vous assurer que la collaboration dispose des informations les plus récentes sur la zone de préparation. Les mises à jour du code ne sont prises en charge qu’avant l’enregistrement de la spécification de code, car c’est à ce moment-là que la version est attribuée et que la somme de contrôle de hachage est calculée.

description: description_text (Facultatif)

Une description de l’artefact (500 caractères maximum).

content_hash: sha256_hash (Facultatif)

Hachage SHA-256 pour la vérification de l’intégrité (64 caractères hexadécimaux).

functions (Nécessaire si aucune procédure n’est définie)

Une liste de définitions UDF ou UDTF.

name identifier

Le nom de la fonction à exposer au modèle appelant. Doit être un identificateur Snowflake valide.

type

Le type de fonction. L’un des rôles suivants : UDF ou UDTF.

language

La langue de la fonction. Actuellement, seulement PYTHON est pris en charge.

runtime_version: python_version (Facultatif)

Version de l’environnement d’exécution Python à utiliser. Versions prises en charge : 3.10 à 3.14.

handler: handler

Le nom de la fonction de gestionnaire dans le code de la fonction à appeler lorsque name est appelé.

arguments (Facultatif)

Arguments de fonction sous forme de liste de paires nom-type. Les types doivent être des types SQL Snowflake valides.

returns: sql_type

Le type de retour. Pour les UDFs, utilisez un type SQL tel que STRING ou FLOAT. Pour UDTFs, utilisez TABLE(column_definitions).

packages (Facultatif)

Une liste des paquets utilisés par ce code. Il peut s’agir de n’importe lesquels de ces paquets Anaconda Python ou de ces paquets API Snowpark. Par exemple : snowflake-snowpark-python, numpy.

imports (Facultatif)

Une liste d’artefacts à importer. Il doit s’agir d’alias de la liste des artefacts de cette spécification.

code_body (Facultatif)

Code Python en ligne. Mutuellement exclusifs avec les importations en zone de préparation. La taille maximale est de 12 MB.

description: description_text (Facultatif)

Une description de la fonction (500 caractères maximum).

procedures (Nécessaire si aucune fonction n’est définie)

Une liste de définitions de procédures stockées. Les champs sont similaires à functions, sauf qu’il n’y a pas de champ type.

Référence API¶

Les procédures suivantes sont utilisées pour gérer les bundles de code personnalisés dans une collaboration :

REGISTER_CODE_SPEC( ['<registry_name>' ,] <code_spec> )

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_CODE_SPEC(
  $$
  api_version: 2.0.0
  spec_type: code_spec
  name: custom_udf
  version: v1
  description: Custom UDF for data normalization

  functions:
    - name: normalize_value
      type: UDF
      language: PYTHON
      runtime_version: "3.10"
      handler: normalize
      arguments:
        - name: value
          type: FLOAT
        - name: min_val
          type: FLOAT
        - name: max_val
          type: FLOAT
      returns: FLOAT
      code_body: |
        def normalize(value, min_val, max_val):
            if max_val == min_val:
                return 0.0
            return (value - min_val) / (max_val - min_val)
  $$
);

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.REGISTER_CODE_SPEC(
  'my_custom_registry',
  $$
  api_version: 2.0.0
  spec_type: code_spec
  name: custom_udf
  version: v1
  description: Custom UDF for data normalization

  functions:
    - name: normalize_value
      type: UDF
      language: PYTHON
      runtime_version: "3.10"
      handler: normalize
      arguments:
        - name: value
          type: FLOAT
        - name: min_val
          type: FLOAT
        - name: max_val
          type: FLOAT
      returns: FLOAT
      code_body: |
        def normalize(value, min_val, max_val):
            if max_val == min_val:
                return 0.0
            return (value - min_val) / (max_val - min_val)
  $$
);

VIEW_REGISTERED_CODE_SPECS( [ '<registry_name>' ] )

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.REGISTRY.VIEW_REGISTERED_CODE_SPECS();

VIEW_CODE_SPECS( <collaboration_name> )

CALL SAMOOHA_BY_SNOWFLAKE_LOCAL_DB.COLLABORATION.VIEW_CODE_SPECS(
  $collaboration_id
);

Exemples de spécifications¶

api_version: 2.0.0
spec_type: code_spec
name: string_utils
version: v1
description: String utility functions

functions:
  - name: clean_string
    type: UDF
    language: PYTHON
    runtime_version: "3.10"
    handler: clean
    arguments:
      - name: input_str
        type: STRING
    returns: STRING
    description: Removes leading/trailing whitespace and converts to lowercase
    code_body: |
      def clean(input_str):
          if input_str is None:
              return None
          return input_str.strip().lower()

  - name: extract_domain
    type: UDF
    language: PYTHON
    runtime_version: "3.10"
    handler: extract
    arguments:
      - name: email
        type: STRING
    returns: STRING
    description: Extracts domain from email address
    code_body: |
      def extract(email):
          if email is None or '@' not in email:
              return None
          return email.split('@')[1]

api_version: 2.0.0
spec_type: code_spec
name: tokenizer
version: v1
description: Text tokenization UDTF

functions:
  - name: tokenize_text
    type: UDTF
    language: PYTHON
    runtime_version: "3.10"
    handler: Tokenizer
    arguments:
      - name: text
        type: STRING
      - name: delimiter
        type: STRING
    returns: TABLE(token STRING, position INTEGER)
    description: Splits text into tokens and returns each with its position
    code_body: |
      class Tokenizer:
          def process(self, text, delimiter):
              if text is None:
                  return
              tokens = text.split(delimiter if delimiter else ' ')
              for i, token in enumerate(tokens):
                  yield (token.strip(), i)

api_version: 2.0.0
spec_type: code_spec
name: ml_scoring
version: v2
description: ML scoring functions using custom library

artifacts:
  - alias: ml_lib
    stage_path: "@MY_DB.PUBLIC.CODE_STAGE/libs/ml_scoring_lib-1.0.0-py3-none-any.whl"
    description: Custom ML scoring library
    content_hash: "a1b2c3d4e5f6..."

functions:
  - name: predict_score
    type: UDF
    language: PYTHON
    runtime_version: "3.10"
    handler: ml_scoring_lib.predictor.predict
    arguments:
      - name: features
        type: ARRAY
    returns: FLOAT
    packages:
      - numpy
      - scikit-learn
    imports:
      - ml_lib
    description: Predicts score using trained ML model

api_version: 2.0.0
spec_type: code_spec
name: data_processor
version: v1
description: Data processing procedures

procedures:
  - name: aggregate_metrics
    language: PYTHON
    runtime_version: "3.10"
    handler: process
    arguments:
      - name: table_name
        type: STRING
      - name: group_column
        type: STRING
    returns: STRING
    packages:
      - snowflake-snowpark-python
    description: Aggregates metrics by specified column
    code_body: |
      def process(session, table_name, group_column):
          df = session.table(table_name)
          result = df.group_by(group_column).count()
          result.write.mode("overwrite").save_as_table("aggregated_results")
          return f"Aggregated {df.count()} rows into aggregated_results"

api_version: 2.0.0
spec_type: code_spec
name: analytics_suite
version: v3
description: Analytics suite with multiple modules

artifacts:
  - alias: utils
    stage_path: "@MY_DB.PUBLIC.CODE_STAGE/analytics/utils.py"
    description: Utility functions
  - alias: transformers
    stage_path: "@MY_DB.PUBLIC.CODE_STAGE/analytics/transformers.py"
    description: Data transformation functions
  - alias: validators
    stage_path: "@MY_DB.PUBLIC.CODE_STAGE/analytics/validators.py"
    description: Validation functions

functions:
  - name: transform_and_validate
    type: UDF
    language: PYTHON
    runtime_version: "3.10"
    handler: transformers.transform_validate
    arguments:
      - name: data
        type: OBJECT
    returns: OBJECT
    imports:
      - utils
      - transformers
      - validators
    description: Transforms and validates input data

Dépannage des bundles de code¶

Erreur:

CodeSpecAlreadyExistsException

Cause:

Spécification de bundle de code avec le même nom et la même version déjà enregistrée.

Solution:

Utilisez une autre version ou mettez à jour la version existante.

Erreur:

SpecValidationError

Cause:

YAML n’est pas conforme au schéma.

Solution:

Vérifiez les champs obligatoires et le format.

Erreur:

CodeSpecStageNotAccessibleError

Cause:

La zone de préparation référencée dans l’artefact n’est pas accessible.

Solution:

Accordez l’accès à la zone de préparation ou vérifiez que la zone de préparation existe.

Erreur:

CodeSpecArtifactNotFoundAtStageError

Cause:

Fichier introuvable dans le chemin de zone de préparation spécifié.

Solution:

Chargez le fichier dans la zone de préparation avant de l’enregistrer.

Erreur:

StageDirectoryNotEnabledError

Cause:

La zone de préparation n’a pas DIRECTORY activé.

Solution:

Activez le répertoire sur la zone de préparation : ALTER STAGE ... SET DIRECTORY = (ENABLE = TRUE)

Erreur:

CodeSpecNotFoundForOwnerException

Cause:

Le modèle fait référence à une spécification de bundle de code non enregistrée.

Solution:

Enregistrez la spécification du bundle de code avant d’enregistrer le modèle.