Exécution de Snowflake SQL avec Snowpark Connect for Spark¶

Pour exécuter les commandes SQL spécifiques à Snowflake, vous pouvez utiliser l’interface SnowflakeSession. Comme avec la méthode spark.sql, les résultats de la requête sont retournés sous forme de DataFrames Spark avec lesquels vous pouvez continuer à appliquer ou à lier des transformations et des actions de DataFrame Spark sur les données résultantes.

Avec la plupart des opérations SQL, vous pouvez utiliser la méthode spark.sql pour exécuter les instructions SQL directement et récupérer les résultats sous forme de DataFrames Spark. Cependant, certaines parties de la syntaxe Snowflake SQL — notamment QUALIFY, CONNECT BY, LATERAL FLATTEN, et les requêtes Time Travel — ne sont pas compatibles avec Spark SQL.

L’exemple suivant indique comment utiliser une SnowflakeSession pour exécuter une commande Snowflake SQL qui inclut la clause CONNECT BY.

import snowflake.snowpark_connect
from snowflake.snowpark_connect.snowflake_session import SnowflakeSession

spark = snowflake.snowpark_connect.server.init_spark_session()
snowflake_session = SnowflakeSession(spark)
result_df = snowflake_session.sql("""
  SELECT
  employee_name,
  manager_name,
  LEVEL
FROM employees
START WITH employee_name = 'Alice'
CONNECT BY PRIOR manager_name = employee_name
""").show()
result_df.limit(1).show()

Vous pouvez également utiliser l’interface SnowflakeSession pour exécuter des directives de configuration spécifiques à Snowflake. Ces directives incluent la définition de paramètres au niveau de la session tels que la base de données, le schéma ou l’entrepôt actif.

L’exemple suivant indique comment utiliser une SnowflakeSession pour définir des paramètres au niveau de la session.

import snowflake.snowpark_connect
from snowflake.snowpark_connect.client import SnowflakeSession

spark = snowflake.snowpark_connect.server.init_spark_session()
snowflake_session = SnowflakeSession(spark)

snowflake_session.use_database("MY_DATABASE")
snowflake_session.use_schema("MY_SCHEMA")
snowflake_session.use_warehouse("MY_WH")
snowflake_session.use_role("PUBLIC")