クリーンルームでカスタム関数をアップロードして実行する¶

概要¶

カスタムPython UDFs および UDTFs をクリーンルームにアップロードし、テンプレートから実行して複雑なデータアクションを実行します。これらのアクションには、単一のステップまたは :doc:`複数のステップフロー </user-guide/cleanrooms/multistep-flows>`の一部として、クエリ内での機械学習やカスタマイズされたデータ操作が含まれます。Pythonは、カスタム UDFs でサポートされている唯一のコーディング言語です。

アップロードされたコードは、Pythonパッケージの承認済みバンドル および Snowpark API からパッケージをインポートして、使用できます。

プロバイダーもコンシューマーも、カスタムPythonコードをクリーンルームにアップロードできますが、プロバイダーとコンシューマーではプロセスが異なります。

このトピックは、カスタムPython UDFs および UDTFs をプロバイダーまたはコンシューマーとしてアップロードし、実行する方法を示しています。

SELECT cleanroom.simple_add({{ price | sqlsafe | int }}, {{ tax | sqlsafe | int }}) ...

SELECT
  cleanroom.simple_add(
    {{ price | sqlsafe | int }}, {{ tax | sqlsafe | int }}
    ) AS TOTAL_ITEM_COST
...

プロバイダー送信コード¶

プロバイダー送信関数は、インラインコードとして、またはSnowflakeステージからアップロードできます。両方の手法はここで説明されています。

アップロードされたコードは、Pythonパッケージの承認済みセット からネイティブにパッケージをインポートして、使用できます。デフォルト以外のパッケージが必要な場合は、クリーンルームでのSnowpark Container Services を使用してコードをホストする必要があります。

自分のコードであっても、アップロードされたコードは表示できません。そのため、必ずクリーンルームにアップロードするものの正確なコピーを追加してください。

プロバイダーがクリーンルームにコードを追加する方法の概要は次のとおりです。

1. プロバイダーは通常の方法でクリーンルームを作成し、構成します。

2. プロバイダーは provider.load_python_into_cleanroom を呼び出してコードをアップロードします。そのプロシージャ内で直接:ref:コードをインラインでアップロードする<label-dcr_upload_code_inline>`か、:ref:`ステージにコードファイルをアップロード <label-dcr_provider_code_from_stage> して、そのプロシージャにステージの場所を提供できます。

   コードには複数の関数を含めることができますが、アップロードごとに公開されるハンドラーは1つだけです。複数の関数をテンプレートに公開するには、provider.load_python_into_cleanroom を呼び出して各ハンドラーをアップロードします。

3. コードのアップロードが成功するたびに、クリーンルームの新しいパッチバージョンが生成されます。その後、新しいパッチ番号を使用して provider.set_default_release_directive を呼び出し、デフォルトバージョンを増やす必要があります。クリーンルームが外部に公開されている場合、コードをインストールする前にセキュリティチェックが実行されます。provider.view_cleanroom_scan_status を呼び出して、デフォルトバージョンをインクリメントする前に、セキュリティチェックに合格したことを確認します。

   • 1つのパッチで複数の関数をアップロードする場合は、コードの一括アップロード が可能です。ただし、一括アップロードでは、アップロードにセキュリティスキャン問題がある場合、問題の原因となったファイルがエラー応答で報告されないため、デバッグがより困難になる可能性があります。

4. コードを呼び出すカスタムテンプレートを作成し、アップロードします。テンプレートは、cleanroom スコープを使用するハンドラー関数、つまり cleanroom.my_function(...) を呼び出す必要があります。

5. コンシューマーは、他のテンプレートと同じ方法でテンプレートを実行します。

   Tip

   コンシューマーがカスタムコードでクリーンルームをインストールする際にマウントエラーが発生した場合、これはコードの構文エラーを示している可能性があります。

このフローを示すコード例は、プロバイダー記述コードの例セクション にあります。

CALL samooha_by_snowflake_local_db.provider.load_python_into_cleanroom(
$cleanroom_name,
'simple_add',                         -- Name used to call the UDF from a template.
['first INTEGER', 'second INTEGER'],  -- Arguments of the UDF, specified as '<variable_name> <SQL type>' pairs.
['numpy', 'pandas'],                  -- Packages imported by the UDF.
'INTEGER',                            -- SQL return type of UDF.
'add_two',                            -- Handler function in your code called when external name is called.
$$
import numpy as np   # Not used, but you can load supported packages.
import pandas as pd

def add_two(first, second):
    return first + second
$$
);

-- Save the following code as reverser.py:
--import numpy as np
--def main(some_string):
--  '''Return the reverse of a string plus a random number 1-10'''
--  return some_string[::-1] + str(np.random.randint(1,10))

-- Get the stage for your clean room.
CALL samooha_by_snowflake_local_db.provider.get_stage_for_python_files($cleanroom_name);

-- Save the file to the stage. Here is how to do it by using the Snowflake CLI
PUT file://~/reverser.py <STAGE_NAME> overwrite=True auto_compress=False;

-- Load the code from the stage into the clean room.
CALL samooha_by_snowflake_local_db.provider.load_python_into_cleanroom(
    $cleanroom_name,
    'reverse', -- Name used to call the function
    ['some_string  STRING'], -- Arguments and SQL types
    ['numpy'],               -- Any required packages
    ['/reverser.py'],        -- Relative path to file on stage
    'STRING',                -- Return type
    'reverser.main'          -- <FILE_NAME>.<FUNCTION_NAME>
);

-- Uploading code, even from a stage, increases the patch number.
CALL samooha_by_snowflake_local_db.provider.set_default_release_directive(
  $cleanroom_name, 'V1_0', <NEW_PATCH_NUMBER>);

-- Upload a template that calls the function.
CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
    $cleanroom_name,
    $udf_template_name,
    $$
    SELECT
      p.status,
      cleanroom.reverse(p.status)
    FROM SAMOOHA_SAMPLE_DATABASE.DEMO.CUSTOMERS AS p
    LIMIT 100;
    $$
);

-- Switch to the consumer account and run the template to see the results.

---- Add custom inline UDF ----
CALL samooha_by_snowflake_local_db.provider.prepare_python_for_cleanroom(
    $cleanroom_name,
    'get_next_status',  -- Name of the UDF. Can be different from the handler.
    ['status VARCHAR'], -- Arguments of the UDF, specified as (variable name, SQL type).
    ['numpy'],          -- Packages needed by UDF.
    [],                 -- When providing the code inline, this is an empty array.
    'VARCHAR',          -- Return type of UDF.
    'get_next_status',  -- Handler.
    $$
import numpy as np
def get_next_status(status):
  """Return the next higher status, or a random status
  if no matching status found or at the top of the list."""

  statuses = ['MEMBER', 'SILVER', 'GOLD', 'PLATINUM', 'DIAMOND']
  try:
    return statuses[statuses.index(status.upper()) + 1]
  except:
    return 'NO MATCH'
    $$
);

---- Add custom inline UDTF. ----
CALL samooha_by_snowflake_local_db.provider.prepare_python_for_cleanroom(
    $cleanroom_name,
    'get_info',  -- Name of the UDTF. Can be different from the handler.
    ['hashed_email VARCHAR', 'days_active INT', 'status VARCHAR', 'income VARCHAR'],   -- Name/Type arguments of the UDTF.
    ['numpy'],         -- Packages used by UDTF.
    [],                -- When providing the code inline, this is an empty array.
    'TABLE(hashed_email VARCHAR, months_active INT, level VARCHAR)',  -- Return type of UDTF.
    'GetSomeVals',     -- Handler class name.
$$
class GetSomeVals:
  def __init__(self):
    self.month_days = 30

  def process(self, hashed_email, days_active, status, income):
    '''Change days into rough months, and also return whether we
    think the user's membership status is lower, higher, or equal to
    what is expected, based on their income.'''

    months_active = days_active // self.month_days
    brackets = ['0-50K', '50K-100K', '100K-250K', '250K+']
    statuses = ['MEMBER', 'SILVER', 'GOLD', 'PLATINUM']
    if(statuses.index(status) < brackets.index(income)):
      level = 'low'
    elif(statuses.index(status) > brackets.index(income)):
      level = 'high'
    else:
      level = 'equal'

    yield(hashed_email, months_active, level)
$$
);

-- Upload all stored procedures.
-- Note the new patch number returned by this procedure. Keep this number for later use.
CALL samooha_by_snowflake_local_db.provider.load_prepared_python_into_cleanroom($cleanroom_name);

-- Set the release directive specified by the last load_python_into_cleanroom call.
CALL samooha_by_snowflake_local_db.provider.set_default_release_directive($cleanroom_name, 'V1_0', <PATCH_NUMBER>);

コンシューマー送信コード¶

コンシューマーがアップロードしたコードはバンドルされ、コンシューマーのテンプレートアップロードフロー を使用したカスタムテンプレートでアップロードされます。アップロードされたコードは、クリーンルームのどのテンプレートからでも呼び出すことができます。

コンシューマーとしてコードをアップロードするには、カスタムテンプレート構文 を理解する必要があります。

コンシューマーがアップロードするコードは、アップロードの権限をリクエストする際に、プロバイダーに表示される可能性があることに注意してください。コンシューマーコードは、プロバイダーまたはコンシューマーがテンプレートを調べるときにも表示されます。

カスタムコンシューマーコードをアップロードする手順の概要は次のとおりです。

1. プロバイダーは標準的な方法でクリーンルームを作成し、コンシューマーを招待します。

2. コンシューマーは標準的な方法でクリーンルームをインストールし、構成します。

3. コンシューマーは、cleanroom 名前空間の中で UDF または UDTF を呼び出すテンプレートを準備します。たとえば、コンシューマー定義の calculate_tax 関数を呼び出す場合、簡潔なテンプレートは次のスニペットのようになります。

   SELECT {{ cleanroom.calculate_tax(p.cost) }} AS Tax FROM my_db.my_sch.sales AS p;
   

   Copy

4. コンシューマーはPythonコードを準備します。後で余計なエスケープが必要にならないよう、コード内では一重引用符（' '）ではなく二重引用符（" "）を使用することをお勧めします。コードは これらのサポートされているPythonライブラリ を参照できます。

5. コンシューマーはPythonコードを consumer.generate_python_request_template に渡します。プロシージャは、カスタム JinjaSQL テンプレートのプレースホルダーと一緒にPythonコードをストアドプロシージャとして返します。$$ を複数行の区切り文字として使用するテンプレートには、複数行の文字列がいくつかあります。

6. コンシューマーは、generate_python_request_template からの出力のテンプレートホルダーを JinjaSQL テンプレートに置き換えます。

7. 結合されたテンプレートでは、一重引用符 \' をエスケープします。これは、複数行のプロシージャ文字列全体の最も外側の区切り文字として一重引用符が使用されるためです。以下は、コンシューマーPythonコードとカスタムテンプレートを含み、文字エスケープされたストアドプロシージャの例です。

     BEGIN
   
     CREATE OR REPLACE FUNCTION CLEANROOM.custom_compare(min_status STRING, max_status STRING, this_status STRING)
     RETURNS boolean
     LANGUAGE PYTHON
     RUNTIME_VERSION = 3.10
     PACKAGES = (\'numpy\')
   
     HANDLER = \'custom_compare\'
     AS $$
     import numpy as np
   
     def custom_compare(min_status:str, max_status:str, this_status:str):
       statuses = [\'MEMBER\', \'SILVER\', \'GOLD\', \'PLATINUM\']
       return ((statuses.index(this_status) >= statuses.index(min_status)) &
               (statuses.index(this_status) <= statuses.index(max_status)))
     $$;
   
     -- Custom template
     LET SQL_TEXT varchar := $$
     SELECT
       c.status,
       c.hashed_email
     FROM IDENTIFIER( {{ my_table[0] }} ) as c
     WHERE cleanroom.custom_compare({{ min_status }}, {{ max_status }}, c.status);
     $$;
   
     LET RES resultset := (EXECUTE IMMEDIATE :SQL_TEXT);
     RETURN TABLE(RES);
   
     END;
   

   Copy

8. コンシューマーは、テンプレートを組み合わせた consumer.create_template_request を呼び出します。template_definition 引数のストアドプロシージャに提供するコードに、二重ドル記号区切り文字（$$...$$）の代わりに一重引用符（' '）を使用します。例:

   CALL samooha_by_snowflake_local_db.consumer.create_template_request(
     $cleanroom_name,
     $template_name,
     '
   BEGIN
   
   -- First, define the Python UDF.
   CREATE OR REPLACE FUNCTION CLEANROOM.custom_compare(min_status STRING, max_status STRING, this_status STRING)
   RETURNS boolean
   LANGUAGE PYTHON
   RUNTIME_VERSION = 3.10
   PACKAGES = (\'numpy\')
   
   HANDLER = \'custom_compare\'
   AS $$
   import numpy as np
   
   def custom_compare(min_status:str, max_status:str, this_status:str):
     statuses = [\'MEMBER\', \'SILVER\', \'GOLD\', \'PLATINUM\']
     return ((statuses.index(this_status) >= statuses.index(min_status)) &
             (statuses.index(this_status) <= statuses.index(max_status)))
       $$;
   
   -- Then define and execute the SQL query.
   LET SQL_TEXT varchar := $$
   SELECT
     c.status,
     c.hashed_email
   FROM IDENTIFIER( {{ my_table[0] }} ) as c
   WHERE cleanroom.custom_compare({{ min_status }}, {{ max_status }}, c.status);
   $$;
   
   -- Execute the query and then return the result.
   LET RES resultset := (EXECUTE IMMEDIATE :SQL_TEXT);
   RETURN TABLE(RES);
   
   END;
   ');
   

   Copy

9. コンシューマーとプロバイダーは標準の コンシューマー定義のテンプレートフロー を使用して続行します。

   1. プロバイダーはテンプレートリクエスト（provider.list_pending_template_requests）を表示してから、approve_template_request を呼び出してリクエストを承認します。リクエストで、プロバイダーはテンプレートとバンドルコードを確認できます。

   2. コンシューマーは、リクエストのステータスを確認します（consumer.list_template_requests）。ステータスが APPROVED になったら、テンプレートを実行します（consumer.run_analysis）。