コラボレーションクリーンルームでのカスタム関数のアップロードと使用¶

概要¶

コラボレーターは、カスタムPythonのUDFsおよびUDTFsをコラボレーションにアップロードできます。コラボレーションのテンプレートは、これらの関数を実行して複雑なデータアクションを実行できます。一般的な使用法には、クエリ内の機械学習やカスタマイズされたデータ操作が含まれます。アップロードしたコードは、`承認済みのPythonパッケージバンドル<https://repo.anaconda.com/pkgs/snowflake/>`_および:ref:`SnowparkAPIパッケージ<label-dcr_snowpark_udf>`からパッケージをインポートして使用できます。

次のセクションでは、カスタム関数をアップロードして使用する方法を説明します。

カスタムコードバンドルの定義と使用¶

カスタム関数をアップロードして使用する方法は次のとおりです。

コード送信者

1. REGISTER_CODE_SPEC。

   コードは仕様内にインラインで記述するか、ステージからリンクできます。

2. テンプレートの``code_specs``配列内でIDによって:ref:コードバンドル仕様を参照するテンプレートを作成します<label-dcr_collaboration_create_and_register_code_bundle_template>。この例に示すように、このフィールドをテンプレートおよびパラメーターフィールドと同列に追加します。

   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. テンプレートを登録してから、テンプレートをコラボレーションにリンクします。

   注釈

   Snowflakeは、アップロードされたコードのセキュリティ問題をスキャンします。セキュリティの問題が見つかった場合、コードとそれを含むテンプレートはコラボレーションに追加されません。

分析実行者：

• ``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] }}
  $$
);

コードバンドル仕様¶

この仕様は、テンプレートから呼び出すことができる1つ以上のコード関数またはプロシージャのバンドルを定義します。

コードバンドル仕様には、関数とプロシージャを合わせて最大5つまで含めることができます。

コードバンドル仕様の識別子には、以下の一般的な要件があります。

• 名前：文字で始まり、英数字とアンダースコアのみを含む、有効な:doc:`Snowflake識別子</sql-reference/identifiers-syntax>`である必要があります。

• 引用符で囲まれた識別子：特殊文字を含む名前では、二重引用符で囲まれた識別子がサポートされています。

• 大文字と小文字の区別：引用符で囲まれていない識別子は大文字と小文字を区別しません。引用符で囲まれた識別子は大文字と小文字を保持します。

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

使用するコラボレーション API のバージョン。2.0.0 である必要があります。

spec_type

仕様タイプ識別子。code_spec である必要があります。

name: identifier

このレジストリ内における、このコードバンドル仕様の一意の名前。最大75文字の有効な:doc:Snowflake識別子</sql-reference/identifiers-syntax>`である必要があります。これは、テンプレートで関数を呼び出すときの最後の名前セグメントとして使用されます。:samp:`cleanroom.{code_spec_name}${function_name}

version: version_id

カスタムバージョン識別子。アンダースコアを含む英数字で、最大20文字である必要があります。

:samp:`description: {description_text}`（オプション）

コードバンドル仕様の説明（最大1,000文字）。

``artifacts``（オプション）

関数またはプロシージャからインポートでき、オプションでハンドラー関数を介して公開できる、ステージングされたファイルまたはパッケージのリスト。仕様ごとに最大5つ。

alias: identifier

インポートでこのアーティファクトを参照するためのエイリアス。この仕様内でこのエイリアスを参照する場合は、:samp:`cleanroom.{spec_name}${alias}`ではなく修飾子のないエイリアス名を使用します。つまり、この仕様内の別の関数を参照するには、修飾子のない関数名を使用します。

stage_path: stage_path

アーティファクトファイルへの完全なステージパス（例：@DB.SCHEMA.STAGE/path/file.whl）。

• **ステージは内部である必要があります。**外部ステージはサポートされていません。

• **ステージではDIRECTORYが有効になっている必要があります。**アーティファクトを含むステージには``DIRECTORY = TRUE``を設定する必要があります。

• ステージパス形式：:samp:`@[{DB}.]SCHEMA.STAGE/path/to/file.ext`形式に従う必要があります。

• パストラバーサルなし：ステージパスに``..``または````を含めることはできません。

• このアーティファクトが存在する必要があります。コードバンドルを登録する際、ファイルが指定されたステージパスに存在している必要があります。

• **ステージではSNOWFLAKE_SSEサーバー側暗号化が有効になっている必要があります。**ステージを作成または変更する場合は、:code:`ENCRYPTION = (TYPE = 'SNOWFLAKE_SSE')`を設定します。

• **ステージングされたコードファイルをプッシュ、削除、または更新する場合**は、コラボレーションにステージからの最新情報が確実に反映されるように、:samp:`ALTER STAGE {stage name} REFRESH`を呼び出す必要があります。コードの更新は、コード仕様を登録する前にのみサポートされます。このときにバージョンが割り当てられ、ハッシュチェックサムが計算されるためです。

:samp:`description: {description_text}`（オプション）

アーティファクトの説明（最大500文字）。

:samp:`content_hash: {sha256_hash}`（オプション）

整合性検証用のSHA-256ハッシュ（64文字の16進数）。

``functions``（プロシージャが定義されていない場合は必須）

UDFまたはUDTF定義のリスト。

name identifier

呼び出しテンプレートに公開する関数名。有効な:doc:`Snowflake識別子</sql-reference/identifiers-syntax>`である必要があります。

type

関数タイプ。UDF の1つ、または UDTF。

language

関数言語。現在のところ、 PYTHON のみサポートされています。

:samp:`runtime_version: {python_version}`（オプション）

使用するPythonランタイムバージョン。サポートされているバージョンは``3.10``から``3.14``です。

handler: handler

``name``が呼び出されたときに呼び出される、関数コード内のハンドラー関数の名前。

``arguments``（オプション）

名前と型のペアのリストとしての関数引数。型は有効なSnowflake SQL型である必要があります。

returns: sql_type

戻り値の型。UDFsの場合は、``STRING``や``FLOAT``などのSQL型を使用します。UDTFs には、 TABLE(column_definitions) を使用します。

``packages``（オプション）

このコードで使用されるパッケージのリスト。これには、これらのAnaconda Pythonパッケージ<https://repo.anaconda.com/pkgs/snowflake/>`_または:ref:`これらのSnowpark APIパッケージ<label-dcr_snowpark_udf>`のいずれかを指定できます。例： ``snowflake-snowpark-python`、 numpy。

``imports``（オプション）

インポートするアーティファクトのリスト。これらは、この仕様のアーティファクトリストのエイリアスである必要があります。

``code_body``（オプション）

インラインのPythonコード。ステージ上のインポートとは排他的です。最大サイズは12MBです。

:samp:`description: {description_text}`（オプション）

関数の説明（最大500文字）。

``procedures``（関数が定義されていない場合は必須）

ストアドプロシージャの定義のリスト。``type``フィールドがない点を除き、フィールドは``functions``と類似しています。

API リファレンス¶

以下の手順は、コラボレーションでカスタムコードバンドルを管理するために使用されます。

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

仕様の例¶

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

コードバンドルのトラブルシューティング¶

エラー:

CodeSpecAlreadyExistsException

原因:

同じ名前とバージョンのコードバンドル仕様が既に登録されています。

解決策:

別のバージョンを使用するか、既存のバージョンを更新します。

エラー:

SpecValidationError

原因:

YAMLはスキーマに準拠していません。

解決策:

必須フィールドと形式を確認してください。

エラー:

CodeSpecStageNotAccessibleError

原因:

アーティファクトで参照されているステージにアクセスできません。

解決策:

ステージへのアクセス権を付与するか、ステージが存在することを確認してください。

エラー:

CodeSpecArtifactNotFoundAtStageError

原因:

指定したステージパスでファイルが見つかりません。

解決策:

登録する前に、ファイルをステージにアップロードしてください。

エラー:

StageDirectoryNotEnabledError

原因:

ステージでDIRECTORYが有効になっていません。

解決策:

ステージでディレクトリを有効にしてください。ALTER STAGE ... SET DIRECTORY = (ENABLE = TRUE)

エラー:

CodeSpecNotFoundForOwnerException

原因:

テンプレートが未登録のコードバンドル仕様を参照しています。

解決策:

テンプレートを登録する前に、コードバンドル仕様を登録してください。