Carregar e usar funções personalizadas no Collaboration Data Clean Rooms¶

Introdução¶

Qualquer colaborador pode carregar UDFs e UDTFs Python personalizadas em uma colaboração. Os modelos na colaboração podem executar essas funções para realizar ações de dados complexas. O uso comum inclui machine learning ou manipulação de dados personalizados em uma consulta. O código carregado pode importar e usar pacotes de um conjunto aprovado de pacotes Python e de pacotes da API Snowpark.

As seções a seguir mostram como carregar e usar uma função personalizada.

Definir e usar pacotes de código personalizados¶

Veja como carregar e usar uma função personalizada:

Quem envia o código:

1. Cria e registra o código chamando REGISTER_CODE_SPEC.

   O código pode estar em linha na especificação ou vinculado de uma área de preparação.

2. Cria um modelo que faz referência à especificação do pacote de código pelo ID na matriz code_specs do modelo. Adicione esse campo como um par de campos de modelo e de parâmetros, conforme mostrado neste exemplo:

   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. Registra e vincula o modelo à colaboração.

   Nota

   O Snowflake verifica se há problemas de segurança no código carregado. Se for encontrado um problema de segurança, o código e o modelo que o contém não serão adicionados à colaboração.

Quem executa a análise:

• Executa o modelo da maneira padrão, chamando 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] }}
  $$
);

Especificação do pacote de código¶

Esta especificação define um pacote de uma ou mais funções ou procedimentos de código que podem ser chamados por um modelo.

Uma especificação de pacote de código tem, no máximo, 5 funções mais procedimentos.

Os identificadores na especificação do pacote de código têm os seguintes requisitos gerais:

• Nomes: devem ser identificadores válidos do Snowflake, que começam com uma letra e contêm apenas caracteres alfanuméricos e sublinhados.

• Identificadores entre aspas: identificadores entre aspas duplas são permitidos para nomes com caracteres especiais.

• Diferenciação entre maiúsculas e minúsculas: identificadores sem aspas não diferenciam maiúsculas de minúsculas. Identificadores entre aspas preservam o uso de maiúsculas e minúsculas.

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

A versão da API do Collaboration utilizada. Deve ser 2.0.0.

spec_type

Identificador do tipo de especificação. Deve ser code_spec.

name: identifier

Um nome exclusivo para esta especificação de pacote de código neste registro. Deve ser um identificador válido do Snowflake com no máximo 75 caracteres. Ele é usado como o segmento do sobrenome ao chamar a função em um modelo: cleanroom.code_spec_name$function_name

version: version_id

Identificador de versão personalizado. Deve ser alfanumérico com sublinhados e, no máximo, 20 caracteres.

description: description_text (opcional)

Uma descrição da especificação do pacote de código (máximo de 1.000 caracteres).

artifacts (opcional)

Uma lista de arquivos ou pacotes preparados que podem ser importados por suas funções ou procedimentos e opcionalmente expostos por meio de funções do manipulador. Máximo de 5 por especificação.

alias: identifier

Um alias para referenciar este artefato nas importações. Ao referenciar este alias nesta especificação, use o nome do alias simples em vez de cleanroom.spec_name$alias; ou seja, use o nome da função simples para fazer referência a outra função nesta especificação.

stage_path: stage_path

Caminho completo da área de preparação para o arquivo de artefato. Por exemplo, @DB.SCHEMA.STAGE/path/file.whl.

• A área de preparação deve ser interna. Áreas de preparação externas não são compatíveis.

• A área de preparação deve ter o DIRECTORY habilitado: a área de preparação que contém artefatos deve ter DIRECTORY = TRUE definido.

• Formato do caminho da área de preparação: deve seguir o formato @[DB.]SCHEMA.STAGE/path/to/file.ext.

• Sem travessia de caminho: os caminhos de área de preparação não podem conter .. ou \.

• Este artefato deve existir: o arquivo deve existir no caminho da área de preparação especificado quando o pacote de código for registrado.

• A área de preparação deve ter a criptografia do lado do servidor SNOWFLAKE_SSE habilitada. Ao criar ou alterar a área de preparação, defina ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE').

• Se você enviar, excluir ou atualizar um arquivo de código preparado, deverá chamar ALTER STAGE stage name REFRESH para garantir que a colaboração tenha as informações mais recentes da área de preparação. As atualizações de código são permitidas apenas antes de você registrar a especificação do código, pois é quando a versão é atribuída e a soma de verificação do hash é calculada.

description: description_text (opcional)

Uma descrição do artefato (máximo de 500 caracteres).

content_hash: sha256_hash (opcional)

Hash SHA-256 para verificação de integridade (64 caracteres hexadecimais).

functions (obrigatório se nenhum procedimento for definido)

Uma lista de definições de UDF ou UDTF.

name identifier

O nome da função a ser exposta ao modelo de chamada. Deve ser um identificador válido do Snowflake.

type

O tipo de função. UDF ou UDTF.

language

A linguagem da função. Atualmente, somente PYTHON é compatível.

runtime_version: python_version (opcional)

Versão de tempo de execução do Python a ser usada. Versões compatíveis: 3.10 a 3.14.

handler: handler

O nome da função do manipulador no código da função a ser chamada quando name é chamado.

arguments (opcional)

Argumentos de função como uma lista de pares nome-tipo. Devem ser tipos SQL válidos do Snowflake.

returns: sql_type

O tipo de retorno. Para UDFs, use um tipo SQL, como STRING ou FLOAT. Para UDTFs, use TABLE(column_definitions):

packages (opcional)

Uma lista de pacotes usados por este código. Pode ser qualquer um destes pacotes Anaconda Python ou destes pacotes da API Snowpark. Por exemplo: snowflake-snowpark-python, numpy.

imports (opcional)

Uma lista de artefatos a serem importados. Eles devem ser aliases da lista de artefatos nesta especificação.

code_body (opcional)

Código Python em linha. Mutualmente exclusivo com importações preparadas. O tamanho máximo é 12 MB.

description: description_text (opcional)

Uma descrição da função (máximo de 500 caracteres).

procedures (obrigatório se nenhuma função for definida)

Uma lista de definições de procedimentos armazenados. Os campos são semelhantes a functions, exceto que não há um campo type.

Referência de API¶

Os procedimentos a seguir são usados para gerenciar pacotes de código personalizados em uma colaboração:

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
);

Exemplos de especificações¶

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

Solução de problemas de pacotes de código¶

Erro:

CodeSpecAlreadyExistsException

Causa:

Especificação do pacote de código com os mesmos nome e versão já registrados.

Solução:

Use uma versão diferente ou atualize a versão existente.

Erro:

SpecValidationError

Causa:

O YAML não está de acordo com o esquema.

Solução:

Verifique os campos e o formato obrigatórios.

Erro:

CodeSpecStageNotAccessibleError

Causa:

A área de preparação referenciada no artefato não está acessível.

Solução:

Conceda acesso à área de preparação ou verifique se ela existe.

Erro:

CodeSpecArtifactNotFoundAtStageError

Causa:

Arquivo não encontrado no caminho da área de preparação especificado.

Solução:

Carregue o arquivo para a área de preparação antes do registro.

Erro:

StageDirectoryNotEnabledError

Causa:

A área de preparação não tem o DIRECTORY habilitado.

Solução:

Habilite o diretório na área de preparação: ALTER STAGE ... SET DIRECTORY = (ENABLE = TRUE)

Erro:

CodeSpecNotFoundForOwnerException

Causa:

O modelo faz referência a uma especificação de pacote de código não registrada.

Solução:

Registre a especificação do pacote de código antes de registrar o modelo.