Conception de flux à plusieurs étapes¶

Vue d’ensemble¶

La plupart des utilisations en clean room consistent à exécuter une seule requête SQL sur une ou plusieurs tables d’une clean room et à afficher les résultats dans la réponse. Cependant, il existe de nombreux cas d’utilisation où vous pourriez vouloir décomposer votre flux en plusieurs étapes, qui peuvent être exécutées de manière séquentielle ou dans n’importe quel ordre, et qui peuvent impliquer l’appel à du code Python pour traiter (ou prétraiter) les données. Il s’agit par exemple d’un flux de machine learning où les données sont entraînées une fois par rapport à un ensemble de données, puis exécutées plusieurs fois par rapport à des données d’entrée variables, soit à l’unité, soit par lots.

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.

• Tables internes : Votre modèle ou vos fonctions internes personnalisées peuvent créer des tables à l’intérieur d’une clean room. Ces tables se comportent comme des tables liées en ce sens qu’elles sont accessibles aux modèles ou au code personnalisé téléchargé. Les tables internes sont utiles pour conserver un état ou des données ; dans l’exemple du machine learning, les données d’apprentissage seraient enregistrées dans une table interne et utilisées par des fonctions internes. L’accès à ces tables n’est possible qu’au moyen de modèles ou de codes téléchargés à l’intérieur de la clean room. Le stockage des données intermédiaires dans des tables internes est beaucoup plus efficace que le passage de grands blocs d’information dans et hors de la clean room à l’aide de modèles.

• Fonctions internes personnalisées : Vous pouvez définir des fonctions personnalisées au sein d’une clean room qui peuvent être appelées par des modèles dans cette clean room. Les fonctions peuvent être définies dans une clean room soit en téléchargeant les UDFs Python ou UDTFs dans la clean room, soit en créant un service de conteneur dans votre clean room qui expose des points de terminaison qui mettent en œuvre des fonctions. Ces fonctions ne peuvent être appelées que par des modèles à l’intérieur de la clean room.

Tables internes de clean room¶

Vous pouvez créer des tables à l’intérieur d’une clean room en utilisant SQL ou Python pour stocker des résultats intermédiaires, ou pour un stockage persistant pour l’utilisateur ou vos fonctions internes (pensez aux données d’entraînement). Ces tables se comportent de la même manière que les tables liées, avec les remarques suivantes :

• Ces tables sont créées à l’aide d’un modèle de clean room ou d’un UDF/UDTF, et n’ont aucun lien avec des tables extérieures.

• Ces tables doivent être créées dans l’espace de noms cleanroom.

• Vous pouvez paramétrer des politiques de lignes et de colonnes sur les tables internes créées manuellement après leur création.

• Si le nom de la table n’est pas statique et que d’autres modèles ou codes doivent y accéder, vous devez renvoyer le nom de la table à l’utilisateur, afin qu’il puisse transmettre le nom de la table dynamique à tout autre modèle devant accéder à cette 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.activation_data_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.

   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. L’UDF Python renvoie le suffixe du nom de la table à l’appelant du modèle.

   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 suffix: {suffix}'
   

   Copy

3. Le modèle de fournisseur accepte un suffixe de nom de table qui lui est transmis et affiche le contenu de cette table.

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

   Copy

4. Le consommateur appelle le modèle en indiquant le suffixe du nom de la table.

   CALL samooha_by_snowflake_local_db.consumer.run_analysis(
     $cleanroom_name,
     'prod_get_results',
     [],
     [],
     object_construct(
         'results_suffix', $result_suffix  -- Table name suffix to identify the results table.
     )
   );
   

   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.