代码捆绑包

Any collaborator can bundle custom Python Procedures, UDFs or UDTFs with collaboration templates. Templates in turn reference the bundled code to perform complex data actions in the collaboration. Common usage includes machine learning or customized data manipulation within a query. Your uploaded code can import and use packages from an approved bundle of Python packages (https://repo.anaconda.com/pkgs/snowflake/) and the Snowpark API.

自定义代码只能通过模板调用，不能直接调用。

Note

代码捆绑包仅支持 Python 编程语言。

以下部分向您展示如何上传和使用代码捆绑包。

实施自定义代码捆绑包

以下是上传和使用代码捆绑包的方法：

代码提交者：

Creates and registers the code by calling REGISTER_CODE_SPEC.

The code can be inline in the spec, or linked from a stage.
Creates a template that references the code bundle spec by ID in the template’s code_specs array. Add this field as a peer of the template and parameters fields as shown in this example:
Registers the template and then links the template into the collaboration.

分析运行器：

Runs the template in the standard way by calling RUN.

Important

在将任何上传的捆绑包部署到 Clean Room 之前，Snowflake 会对其执行安全检查。如果安全检查失败，该模板及其捆绑代码将不会被部署，也无法使用。

要确认带有代码捆绑包的模板已部署并可供使用，请执行以下步骤：

查找您尝试部署代码捆绑包的洁净室应用程序的名称：
SHOW APPLICATIONS LIKE 'SFDCR_<collaboration name>';
Check the upgrade_state value in the DESCRIBE APPLICATION response. When the upgrade state is COMPLETE, the security checks have passed and the new template and bundle are available to use. Pass in the application name returned by the command in the previous step using SQL like the following example: SQL code:
DESCRIBE APPLICATION <application name>

创建并注册代码包规范

上传自定义代码的第一步是创建并注册代码包规范。

Custom functions are defined in a YAML code bundle spec. Each code bundle exposes one or more functions that can be called by a template. The code bundle spec can either include the code in the spec inline, or link to code that lives on a Snowflake stage.

A collaborator registers a spec by calling REGISTRY.REGISTER_CODE_SPEC, which returns the bundle ID.

After the template that references the code bundle is linked into the collaboration, that code bundle is visible to anyone in the collaboration who can access a template that links the code bundle. Call VIEW_CODE_SPECS to list accessible code bundles in a collaboration.

Anyone who can see a code bundle in a collaboration can see and use it in their own templates in that collaboration. Any inline code can be viewed by any member of the collaboration, but staged artifact code can not be viewed by collaborators. Collaborators need to ensure that the content_hash of the referenced artifacts match for code integrity verification.

The following code bundle spec that exposes a single Python UDF called normalize_value, which calls the normalize function defined in that 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
  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
  $$
);

创建并注册调用模板

After the code spec is registered, the collaborator then registers a template that uses this code bundle. To use a code bundle, add the bundle spec ID in the template’s code_specs field. Adding this template into the collaboration will also cause the code bundled to be available in the collaboration.

A template calls a custom function using the syntax cleanroom.spec_name$function_name. Note the literal . and $ name scoping marks.

Note

请使用规范名称（而不是规范 ID）来引用模板中的函数。这样您就可以快速更新代码捆绑包的版本，而无需更改模板中对该代码捆绑包的所有引用。

In the following example, a template uses function normalize_value from the code bundle custom_udf:

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(100)  -- Calls the UDF.
      AS normalized
        FROM {{ source_tables[0] }}
  $$
);

将模板添加到协作

Add the template that calls your function to the collaboration in the standard way. For more information, see Templates.

当调用模板被添加到协作时，Snowflake 会进行验证并上传到协作中。以下示例展示了向现有协作添加模板的请求：

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

Note

使用代码捆绑包安装模板会触发 Snowflake 安全检查，并发布底层 Clean Room 的新补丁。在此过程完成且补丁安装成功之前，该模板将不可用、也无法使用。

要查看补丁安装进度，请执行以下步骤：

Find the name of the clean room application. Typically, this will be SFDCR_<clean room name>, but you can search to be sure:
```
-- Find the exact name of the clean room application.
SHOW APPLICATIONS LIKE 'SFDCR_%';
```
Check the status of the patch install. Wait for upgrade_state is COMPLETE in the following query:
```
DESCRIBE APPLICATION SFDCR_<application name>;
```

代码版本管理

在账户的所有注册表中，每个注册的代码规范必须具有唯一的名称 + 版本。模板会加载特定名称和版本的代码规范。如果要创建或使用新版本的代码，则必须提交新版本的模板，并在 code_specs 字段中引用新的代码版本。您不需要更改模板正文。例如：

第 1 步： 使用代码包的版本 1：

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(100)  -- Calls the UDF.
      AS normalized
        FROM {{ source_tables[0] }}
  $$
);

第 2 步： 更新并注册代码包的新版本，然后更新模板以使用新版本：

请注意，函数名称不包含版本，因此在上传函数的新版本时，无需更改模板正文中的调用代码。

示例规范

带有代码正文的内联 UDF¶

一个带有内联 Python 代码的简单 UDF：

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]

UDTF（用户定义的表函数）¶

此示例 YAML 定义了一个返回多行的 UDTF：

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)

带有 wheel 包的暂存工件¶

Be sure to read the stage_path documentation requirements for linking to staged code in your code spec.

此示例 YAML 使用暂存的 Python wheel 包：

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

存储过程

此示例 YAML 定义了用于数据处理的存储过程：

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"

作为暂存工件的多个 Python 文件¶

Be sure to read the stage_path documentation requirements for linking to staged code in your code spec.

此示例 YAML 使用多个暂存的 Python 源文件：

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