Conception de flux à plusieurs étapes¶

Vue d’ensemble¶

Most clean room usage involves running a single SQL query against one or more tables in a clean room and displaying the results in the response. However, there are many use cases where you might want to break up your flow into several steps, which can be run sequentially or individually, and can involve calling Python code to process (or pre-process) data. Examples include a machine learning flow where the model is trained once against a data set and then run multiple times against varying input data, either singly or in batches.

Clean rooms disposede plusieurs mécanismes pour permettre ces scénarios avancés :

• Chaînes de modèles : Une chaîne de modèles exécute un ensemble de modèles dans un ordre spécifique, en utilisant la sortie de chaque modèle comme entrée du modèle suivant. L’entrée du premier modèle de la chaîne est fournie par l’utilisateur ; la sortie du dernier modèle de la chaîne est renvoyée à l’utilisateur.

• Internal tables: Your template or custom internal functions can create persistent tables within a clean room. These tables behave like linked tables in that they are accessible to templates or custom uploaded code. Internal tables are useful for maintaining state or data; in the machine learning example, the training data is saved in an internal table that is used by internal functions. Just as with linked tables, these tables can be accessed only by templates or uploaded code inside the clean room. Storing intermediary data in internal tables is more efficient than passing large blocks of information into and out of the clean room using templates.

• Fonctions internes personnalisées : vous pouvez définir des fonctions personnalisées dans une salle blanche qui peuvent être appelées par des modèles de cette salle blanche. Les fonctions peuvent être définies dans une salle blanche en chargeant des UDFs ou des UDTFs Python dans la salle blanche, ou en créant un service de conteneurs dans votre salle blanche qui expose les points de terminaison qui mettent en œuvre des fonctions. Ces fonctions peuvent uniquement appelées par des modèles au sein de la salle blanche.

Tables internes de clean room¶

You can create tables inside a clean room using SQL or Python to store intermediary results, or for persistent storage for the user or your internal functions (for example, to save training data that is used for multiple runs). These tables behave the same as linked tables, with the following notes:

• Internal tables are created using a clean room template or a UDF/UDTF, and have no linkage to outside tables.

• Internal tables are created in the cleanroom namespace.

• You can set row and column policies on internal tables after you create them.

• If the table name is dynamic, and the table is accessed by other templates or code, return the name of the table to the user so the user can pass the dynamic table name to any other templates that need to access that table.

Voici quelques exemples de création d’une table interne :

CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
  $cleanroom_name,
  $template_name,
  $$
  BEGIN
    CREATE OR REPLACE TABLE cleanroom.analysis_results AS
      SELECT count(*) AS ITEM_COUNT, c.status, c.age_band
      FROM IDENTIFIER({{ my_table[0] }}) AS c
      JOIN IDENTIFIER({{ source_table[0] }}) AS p
      ON {{ c_join_col | sqlsafe | activation_policy }} = {{ p_join_col | sqlsafe | activation_policy }}
      GROUP BY c.status, c.age_band
      ORDER BY c.age_band;
      RETURN 'analysis_results';
  END;
  $$);

# Snippet of Python UDF to save results to an internal table.
table_name = f'cleanroom.results'

session.sql(f"""
CREATE OR REPLACE TABLE {table_name} AS (
  WITH joint_data AS (
    SELECT
        date,
        p.hashed_email AS hem,
        impression_id
    FROM {source_table} p
  )
  SELECT
    date,
    COUNT(DISTINCT hem) AS reach,
    COUNT(DISTINCT impression_id) AS num_impressions
  FROM joint_data
  GROUP BY date
  ORDER BY date
);
""").collect()

# Snippet of container services Python code to create an internal results table.
# 'cleanroom' table name prefix is added using the schema parameter when the table is created.
@app.post("/score")
def score():

... omitted content ...

df = pd.DataFrame({
    "ID": ids,
    "SCORE": scores
})
table = "LOOKALIKE_RESULTS"
session.write_pandas(df, table, schema="CLEANROOM", auto_create_table=True, overwrite=True)

end_time = time.perf_counter()
execution_time = end_time - start_time
response = make_json_response([[0, {"results_table": table, "size": len(ids), "execution_time": round(execution_time, 2)}]])
return response

Lorsque vous générez une table interne à laquelle le modèle ou le code doit accéder, vous pouvez soit utiliser un nom de table constant, soit nommer la table de manière dynamique et renvoyer le nom de la table à l’utilisateur, qui transmet ensuite le nom de la table à la fonction de résultats.

Voici un exemple de table nommée dynamiquement utilisée pour stocker des résultats. L’utilisateur fait deux appels : l’un pour générer les données et obtenir le nom de la table, l’autre pour voir les résultats.

1. Le modèle de fournisseur appelle reach_impression_regression UDF pour traiter les données (le préfixe cleanroom indique qu’il s’agit d’un UDF). L’UDF renvoie le nom du préfixe de la table interne au modèle, qui le renvoie à l’appelant.

   -- This template calls a UDF uploaded by a collaborator.
   -- The UDF takes two input tables as parameters.
   CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
     $cleanroom_name,
     'prod_calculate_regression',
     $$
     CALL cleanroom.reach_impression_regression({{ source_table[0] }}, {{ my_table[0] | default('NONE') }});
     $$
   );
   

   Copy

2. The Python UDF generates the internal table and returns the generated table name to the template caller.

   def main(session, source_table, my_table):
     ...
     table = f'results_{suffix}'.upper()
     retval_df = session.write_pandas(regression_output, table,  schema = 'CLEANROOM', auto_create_table = True)
   
     return f'Done, results have been written to the following table: {table}'
   

   Copy

3. The provider template accepts a table name passed in and displays the contents of that table. Note how the table is always accessed from the cleanroom namespace.

   CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
           $cleanroom_name,
           'prod_get_results',
           $$
             SELECT * FROM cleanroom.{{ results_table | sqlsafe }};
           $$
   );
   

   Copy

4. The consumer calls the template, passing in the table name.

   CALL samooha_by_snowflake_local_db.consumer.run_analysis(
     $cleanroom_name,
     'prod_get_results',
     [],
     [],
     object_construct(
         'results_table', $table_name
     )
   );
   

   Copy

Déclencher des fonctions personnalisées¶

Les fonctions personnalisées peuvent être appelées soit par des modèles, soit par du code (UDFs, UDTFs, ou points de terminaison de service conteneur) dans la clean room. Les fonctions téléchargées par un collaborateur peuvent être accédées par des modèles ou du code provenant de n’importe quel autre collaborateur.

Les fonctions Clean room doivent toujours être appelées avec le champ d’application de l’espace de noms approprié :

• cleanroom.function_name lors de l’appel d’une fonction personnalisée UDF/UDTF

• service_functions.function_name lors de l’appel d’une fonction exposée en tant que fonction intégrée du service de conteneurs Snowpark.

Voici des exemples d’appel d’un UDF personnalisé et d’un point de terminaison de service conteneur personnalisé à partir d’un modèle :

-- Template to generate results. Calls the UDF 'my_function', which
-- generates a results table inside the clean room called 'results'.
CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
  $cleanroom_name,
  'generate_results_template',
  $$
  CALL cleanroom.my_function({{ source_table[0] }}, {{ my_table[0] | default('NONE') }});
  $$
);

-- Template to trigger training data generation.
CALL samooha_by_snowflake_local_db.provider.add_custom_sql_template(
  $cleanroom_name,
  'lal_train',
  $$
  SELECT service_functions.my_func(
            {{ source_table[0] }},
            {{ provider_join_col }},
            {{ my_table[0] }},
            {{ consumer_join_col }},
            {{ dimensions | sqlsafe }},
            {{ filter_clause }}
          ) AS train_result;
$$

Modèles courants de flux à plusieurs étapes¶

L’exemple d’API Snowpark traite les données, génère des tables intermédiaires, puis une table de résultats avec un appel de modèle, avant d’exposer directement les résultats par un second appel de modèle.

L’exemple de service conteneur Snowpark crée des données de formation avec un appel de modèle et stocke les données de formation dans une table interne. Un second modèle analyse l’entrée de l’utilisateur par rapport aux données de formation stockées.