Utilisation de DataFrames dans Snowpark Python¶

Dans Snowpark, le principal moyen par lequel vous interrogez et traitez les données est un DataFrame. Cette rubrique explique comment travailler avec des DataFrames.

Pour récupérer et manipuler des données, vous utilisez la classe DataFrame . Un DataFrame représente un ensemble de données relationnelles qui est évalué de façon « lazy » : il ne s’exécute que lorsqu’une action spécifique est déclenchée. En un sens, un DataFrame est comme une requête qui doit être évaluée afin d’extraire des données.

Pour récupérer des données dans un DataFrame :

Construit un DataFrame, en spécifiant la source des données pour l’ensemble de données.

Par exemple, vous pouvez créer un DataFrame pour contenir les données d’une table, d’un fichier CSV externe, de données locales ou de l’exécution d’une instruction SQL.
Spécifier comment l’ensemble de données dans le DataFrame doit être transformé.

Par exemple, vous pouvez spécifier quelles colonnes doivent être sélectionnées, comment les lignes doivent être filtrées, comment les résultats doivent être triés et regroupés, etc.
Exécuter l’instruction pour récupérer les données dans le DataFrame.

Pour récupérer les données dans le DataFrame, vous devez appeler une méthode qui exécute une action (par exemple, la méthode collect()).

Les sections suivantes expliquent ces étapes plus en détail.

Configuration des exemples pour cette section¶

Certains des exemples de cette section utilisent un DataFrame pour interroger une table nommée sample_product_data. Si vous voulez exécuter ces exemples, vous pouvez créer cette table et la remplir avec des données en exécutant les instructions SQL suivantes.

Vous pouvez exécuter les instructions SQL à l’aide de Snowpark Python :

session.sql('CREATE OR REPLACE TABLE sample_product_data (id INT, parent_id INT, category_id INT, name VARCHAR, serial_number VARCHAR, key INT, "3rd" INT)').collect()

[Row(status='Table SAMPLE_PRODUCT_DATA successfully created.')]

session.sql("""
INSERT INTO sample_product_data VALUES
(1, 0, 5, 'Product 1', 'prod-1', 1, 10),
(2, 1, 5, 'Product 1A', 'prod-1-A', 1, 20),
(3, 1, 5, 'Product 1B', 'prod-1-B', 1, 30),
(4, 0, 10, 'Product 2', 'prod-2', 2, 40),
(5, 4, 10, 'Product 2A', 'prod-2-A', 2, 50),
(6, 4, 10, 'Product 2B', 'prod-2-B', 2, 60),
(7, 0, 20, 'Product 3', 'prod-3', 3, 70),
(8, 7, 20, 'Product 3A', 'prod-3-A', 3, 80),
(9, 7, 20, 'Product 3B', 'prod-3-B', 3, 90),
(10, 0, 50, 'Product 4', 'prod-4', 4, 100),
(11, 10, 50, 'Product 4A', 'prod-4-A', 4, 100),
(12, 10, 50, 'Product 4B', 'prod-4-B', 4, 100)
""").collect()

[Row(number of rows inserted=12)]

Pour vérifier que la table a été créée, exécutez :

session.sql("SELECT count(*) FROM sample_product_data").collect()

[Row(COUNT(*)=12)]

Configuration des exemples dans une feuille de calcul Python¶

Pour configurer et exécuter ces exemples dans une feuille de calcul Python, créez la table d’exemple et configurez votre feuille de calcul Python.

Créez une feuille de calcul SQL et exécutez ce qui suit :

CREATE OR REPLACE TABLE sample_product_data
  (id INT, parent_id INT, category_id INT, name VARCHAR, serial_number VARCHAR, key INT, "3rd" INT);

INSERT INTO sample_product_data VALUES
  (1, 0, 5, 'Product 1', 'prod-1', 1, 10),
  (2, 1, 5, 'Product 1A', 'prod-1-A', 1, 20),
  (3, 1, 5, 'Product 1B', 'prod-1-B', 1, 30),
  (4, 0, 10, 'Product 2', 'prod-2', 2, 40),
  (5, 4, 10, 'Product 2A', 'prod-2-A', 2, 50),
  (6, 4, 10, 'Product 2B', 'prod-2-B', 2, 60),
  (7, 0, 20, 'Product 3', 'prod-3', 3, 70),
  (8, 7, 20, 'Product 3A', 'prod-3-A', 3, 80),
  (9, 7, 20, 'Product 3B', 'prod-3-B', 3, 90),
  (10, 0, 50, 'Product 4', 'prod-4', 4, 100),
  (11, 10, 50, 'Product 4A', 'prod-4-A', 4, 100),
  (12, 10, 50, 'Product 4B', 'prod-4-B', 4, 100);

SELECT count(*) FROM sample_product_data;

Créez une feuille de calcul Python, en définissant le même contexte de base de données et de schéma que la feuille de calcul SQL que vous avez utilisée pour créer la table sample_product_data.

Si vous souhaitez utiliser les exemples de cette rubrique dans une feuille de calcul Python, utilisez l’exemple dans la fonction de gestionnaire (par exemple main) et utilisez l’objet Session qui est transmis à la fonction pour créer des DataFrames.

Par exemple, appelez la méthode table de l’objet session pour créer un DataFrame pour une table :

import snowflake.snowpark as snowpark
from snowflake.snowpark.functions import col

def main(session: snowpark.Session):
  df_table = session.table("sample_product_data")

Pour examiner la sortie produite par la fonction, par exemple en appelant la méthode show de l’objet DataFrame, utilisez l’onglet Output.

Pour examiner la valeur renvoyée par la fonction, choisissez le type de données de la valeur renvoyée par Settings » Return type, et utilisez l’onglet Results :

Si votre fonction renvoie un DataFrame, utilisez le type de retour par défaut de Table.
Si votre fonction renvoie la list de Row à partir de la méthode collect d’un objet DataFrame, utilisez Variant pour le type de retour.
Si votre fonction renvoie une autre valeur pouvant être convertie en chaîne ou si votre fonction ne renvoie pas de valeur, utilisez String comme type de retour.

Reportez-vous à Exécution de feuilles de calcul Python pour plus de détails.

Construire un DataFrame¶

Pour construire un DataFrame, vous pouvez utiliser les méthodes et les propriétés de la classe Session. Chacune des méthodes suivantes construit un DataFrame à partir d’un type différent de source de données.

Vous pouvez exécuter ces exemples dans votre environnement de développement local ou les appeler dans la fonction main définie dans une feuille de calcul Python.

Pour créer un DataFrame à partir des données d’une table, d’une vue ou d’un flux, appelez la méthode table :

# Create a DataFrame from the data in the "sample_product_data" table.
df_table = session.table("sample_product_data")

# To print out the first 10 rows, call df_table.show()

Pour créer un DataFrame à partir des valeurs spécifiées, appelez la méthode create_dataframe :

# Create a DataFrame with one column named a from specified values.
df1 = session.create_dataframe([1, 2, 3, 4]).to_df("a")
df1.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df1

-------
|"A"  |
-------
|1    |
|2    |
|3    |
|4    |
-------

Créez un DataFrame avec quatre colonnes « a », « b », « c » et « d » :

# Create a DataFrame with 4 columns, "a", "b", "c" and "d".
df2 = session.create_dataframe([[1, 2, 3, 4]], schema=["a", "b", "c", "d"])
df2.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df2

-------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------

Créez un autre DataFrame avec quatre colonnes « a », « b », « c » et « d » :

# Create another DataFrame with 4 columns, "a", "b", "c" and "d".
from snowflake.snowpark import Row
df3 = session.create_dataframe([Row(a=1, b=2, c=3, d=4)])
df3.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df3

-------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------

Créez un DataFrame et spécifiez un schéma :

# Create a DataFrame and specify a schema
from snowflake.snowpark.types import IntegerType, StringType, StructType, StructField
schema = StructType([StructField("a", IntegerType()), StructField("b", StringType())])
df4 = session.create_dataframe([[1, "snow"], [3, "flake"]], schema)
df4.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df4

---------------
|"A"  |"B"    |
---------------
|1    |snow   |
|3    |flake  |
---------------

Pour créer un DataFrame contenant une plage de valeurs, appelez la méthode range :

# Create a DataFrame from a range
# The DataFrame contains rows with values 1, 3, 5, 7, and 9 respectively.
df_range = session.range(1, 10, 2).to_df("a")
df_range.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df_range

-------
|"A"  |
-------
|1    |
|3    |
|5    |
|7    |
|9    |
-------

Pour créer un DataFrame qui contiendra les données d’un fichier dans une zone de préparation, utilisez la propriété read pour obtenir un objet DataFrameReader. Dans l’objet DataFrameReader appelez la méthode correspondant au format des données du fichier :

from snowflake.snowpark.types import StructType, StructField, StringType, IntegerType

# Create DataFrames from data in a stage.
df_json = session.read.json("@my_stage2/data1.json")
df_catalog = session.read.schema(StructType([StructField("name", StringType()), StructField("age", IntegerType())])).csv("@stage/some_dir")

Pour créer un DataFrame qui contiendra les résultats d’une requête SQL, appelez la méthode sql :

# Create a DataFrame from a SQL query
df_sql = session.sql("SELECT name from sample_product_data")
df_sql.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
# return df_sql

--------------
|"NAME"      |
--------------
|Product 1   |
|Product 1A  |
|Product 1B  |
|Product 2   |
|Product 2A  |
|Product 2B  |
|Product 3   |
|Product 3A  |
|Product 3B  |
|Product 4   |
--------------

Il est possible d’utiliser la méthode sql pour exécuter des instructions SELECT qui récupèrent des données à partir de tables et de fichiers en zone de préparation, mais l’utilisation de la méthode table et de la propriété read offre une meilleure mise en évidence syntaxique, une mise en évidence des erreurs et remplissage automatique et intelligent du code dans des outils de développement.

Spécifier comment l’ensemble de données doit être transformé¶

Pour spécifier les colonnes à sélectionner et comment filtrer, trier, grouper, etc. les résultats, appelez les méthodes DataFrame qui transforment l’ensemble de données. Pour identifier les colonnes dans ces méthodes, utilisez la fonction col ou une expression qui évalue une colonne. Reportez-vous à Spécification des colonnes et des expressions.

Par exemple :

Pour spécifier les lignes à renvoyer, appelez la méthode filter :

# Import the col function from the functions module.
# Python worksheets import this function by default
from snowflake.snowpark.functions import col

# Create a DataFrame for the rows with the ID 1
# in the "sample_product_data" table.

# This example uses the == operator of the Column object to perform an
# equality check.
df = session.table("sample_product_data").filter(col("id") == 1)
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df

------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"     |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
------------------------------------------------------------------------------------
|1     |0            |5              |Product 1  |prod-1           |1      |10     |
------------------------------------------------------------------------------------

Pour spécifier les colonnes qui doivent être sélectionnées, appelez la méthode select :

# Import the col function from the functions module.
from snowflake.snowpark.functions import col

# Create a DataFrame that contains the id, name, and serial_number
# columns in the "sample_product_data" table.
df = session.table("sample_product_data").select(col("id"), col("name"), col("serial_number"))
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df

---------------------------------------
|"ID"  |"NAME"      |"SERIAL_NUMBER"  |
---------------------------------------
|1     |Product 1   |prod-1           |
|2     |Product 1A  |prod-1-A         |
|3     |Product 1B  |prod-1-B         |
|4     |Product 2   |prod-2           |
|5     |Product 2A  |prod-2-A         |
|6     |Product 2B  |prod-2-B         |
|7     |Product 3   |prod-3           |
|8     |Product 3A  |prod-3-A         |
|9     |Product 3B  |prod-3-B         |
|10    |Product 4   |prod-4           |
---------------------------------------

Vous pouvez également référencer les colonnes comme ceci :

# Import the col function from the functions module.
from snowflake.snowpark.functions import col

df_product_info = session.table("sample_product_data")
df1 = df_product_info.select(df_product_info["id"], df_product_info["name"], df_product_info["serial_number"])
df2 = df_product_info.select(df_product_info.id, df_product_info.name, df_product_info.serial_number)
df3 = df_product_info.select("id", "name", "serial_number")

Chaque méthode renvoie un nouvel objet DataFrame qui a été transformé. La méthode n’affecte pas l’objet DataFrame d’origine. Si vous souhaitez appliquer plusieurs transformations, vous pouvez enchaîner les appels de méthode, en appelant chaque méthode de transformation suivante sur le nouvel objet DataFrame renvoyé par l’appel de méthode précédent.

Ces méthodes de transformation spécifient comment construire l’instruction SQL et ne récupèrent pas les données de la base de données Snowflake. Les méthodes d’action décrites dans Exécution d’une action pour évaluer un DataFrame effectuent la récupération des données.

Joindre des DataFrames¶

Pour joindre des objets DataFrame, il faut appeler la méthode join :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# Create a DataFrame that joins the two DataFrames
# on the column named "key".
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].as_("key"), "value1", "value2").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].as_("key"), "value1", "value2")

-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Si les deux DataFrames ont la même colonne à joindre, vous pouvez utiliser l’exemple de syntaxe suivant :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# If both dataframes have the same column "key", the following is more convenient.
df_lhs.join(df_rhs, ["key"]).show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, ["key"])

-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Vous pouvez également utiliser l’opérateur & pour connecter des expressions de jointure :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
# Use & operator connect join expression. '|' and ~ are similar.
df_joined_multi_column = df_lhs.join(df_rhs, (df_lhs.col("key") == df_rhs.col("key")) & (df_lhs.col("value1") < df_rhs.col("value2"))).select(df_lhs["key"].as_("key"), "value1", "value2")
df_joined_multi_column.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined_multi_column

-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------

Si vous souhaitez effectuer une auto-jointure, vous devez copier le DataFrame :

# copy the DataFrame if you want to do a self-join
from copy import copy

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs_copied = copy(df_lhs)
df_self_joined = df_lhs.join(df_lhs_copied, (df_lhs.col("key") == df_lhs_copied.col("key")) & (df_lhs.col("value1") == df_lhs_copied.col("value1")))

Lorsqu’il y a des colonnes qui se chevauchent dans les DataFrames, Snowpark ajoute un préfixe généré aléatoirement aux colonnes dans le résultat de la jointure :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"))

-----------------------------------------------------
|"l_av5t_KEY"  |"VALUE1"  |"r_1p6k_KEY"  |"VALUE2"  |
-----------------------------------------------------
|a             |1         |a             |3         |
|b             |2         |b             |4         |
-----------------------------------------------------

Vous pouvez renommer les colonnes qui se chevauchent à l’aide de Column.alias :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].alias("key1"), df_rhs["key"].alias("key2"), "value1", "value2").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs["key"].alias("key1"), df_rhs["key"].alias("key2"), "value1", "value2")

-----------------------------------------
|"KEY1"  |"KEY2"  |"VALUE1"  |"VALUE2"  |
-----------------------------------------
|a       |a       |1         |3         |
|b       |b       |2         |4         |
-----------------------------------------

Pour éviter les préfixes aléatoires, vous pouvez aussi spécifier un suffixe à ajouter aux colonnes qui se chevauchent :

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value1"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value2"])
df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"), lsuffix="_left", rsuffix="_right").show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"), lsuffix="_left", rsuffix="_right")

--------------------------------------------------
|"KEY_LEFT"  |"VALUE1"  |"KEY_RIGHT"  |"VALUE2"  |
--------------------------------------------------
|a           |1         |a            |3         |
|b           |2         |b            |4         |
--------------------------------------------------

Ces exemples utilisent DataFrame.col pour spécifier les colonnes à utiliser dans la jointure. Reportez-vous à Spécification des colonnes et des expressions pour plus de façons de spécifier des colonnes.

Si vous devez joindre une table avec elle-même sur différentes colonnes, vous ne pouvez pas effectuer l’auto-jointure avec un seul DataFrame. Les exemples suivants utilisent un seul DataFrame pour effectuer une auto-jointure, qui échoue parce que les expressions de colonne pour "id" sont présentes dans les côtés gauche et droit de la jointure :

from snowflake.snowpark.exceptions import SnowparkJoinException

df = session.table("sample_product_data")
# This fails because columns named "id" and "parent_id"
# are in the left and right DataFrames in the join.
try:
  df_joined = df.join(df, col("id") == col("parent_id")) # fails
except SnowparkJoinException as e:
  print(e.message)

You cannot join a DataFrame with itself because the column references cannot be resolved correctly. Instead, create a copy of the DataFrame with copy.copy(), and join the DataFrame with this copy.

# This fails because columns named "id" and "parent_id"
# are in the left and right DataFrames in the join.
try:
  df_joined = df.join(df, df["id"] == df["parent_id"])   # fails
except SnowparkJoinException as e:
  print(e.message)

You cannot join a DataFrame with itself because the column references cannot be resolved correctly. Instead, create a copy of the DataFrame with copy.copy(), and join the DataFrame with this copy.

Au lieu de cela, utilisez la méthode copy() Python intégrée pour créer un clone de l’objet DataFrame, et utilisez les deux objets DataFrame pour effectuer la jointure :

from copy import copy

# Create a DataFrame object for the "sample_product_data" table for the left-hand side of the join.
df_lhs = session.table("sample_product_data")
# Clone the DataFrame object to use as the right-hand side of the join.
df_rhs = copy(df_lhs)

# Create a DataFrame that joins the two DataFrames
# for the "sample_product_data" table on the
# "id" and "parent_id" columns.
df_joined = df_lhs.join(df_rhs, df_lhs.col("id") == df_rhs.col("parent_id"))
df_joined.count()

Spécification des colonnes et des expressions¶

Lorsque vous appelez ces méthodes de transformation, vous pouvez avoir besoin de spécifier des colonnes ou des expressions qui utilisent des colonnes. Par exemple, lorsque vous appelez la méthode select , vous devez spécifier les colonnes à sélectionner.

Pour faire référence à une colonne, créez un objet Column en appelant la fonction col dans le module snowflake.snowpark.functions.

# Import the col function from the functions module.
from snowflake.snowpark.functions import col

df_product_info = session.table("sample_product_data").select(col("id"), col("name"))
df_product_info.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_product_info

---------------------
|"ID"  |"NAME"      |
---------------------
|1     |Product 1   |
|2     |Product 1A  |
|3     |Product 1B  |
|4     |Product 2   |
|5     |Product 2A  |
|6     |Product 2B  |
|7     |Product 3   |
|8     |Product 3A  |
|9     |Product 3B  |
|10    |Product 4   |
---------------------

Note

Pour créer un objet Column pour un littéral, reportez-vous à Utilisation de littéraux comme objets de colonne.

Lorsque vous spécifiez un filtre, une projection, une condition de jointure, etc., vous pouvez utiliser des objets Column dans une expression. Par exemple :

Vous pouvez utiliser des objets Column avec la méthode filter pour spécifier une condition de filtrage :

# Specify the equivalent of "WHERE id = 20"
# in a SQL SELECT statement.
df_filtered = df.filter(col("id") == 20)

df = session.create_dataframe([[1, 3], [2, 10]], schema=["a", "b"])
# Specify the equivalent of "WHERE a + b < 10"
# in a SQL SELECT statement.
df_filtered = df.filter((col("a") + col("b")) < 10)
df_filtered.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_filtered

-------------
|"A"  |"B"  |
-------------
|1    |3    |
-------------

Vous pouvez utiliser les objets Column avec la méthode select pour définir un alias :

df = session.create_dataframe([[1, 3], [2, 10]], schema=["a", "b"])
# Specify the equivalent of "SELECT b * 10 AS c"
# in a SQL SELECT statement.
df_selected = df.select((col("b") * 10).as_("c"))
df_selected.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_selected

-------
|"C"  |
-------
|30   |
|100  |
-------

Vous pouvez utiliser les objets Column avec la méthode join pour définir une condition de jointure :

dfX = session.create_dataframe([[1], [2]], schema=["a_in_X"])
dfY = session.create_dataframe([[1], [3]], schema=["b_in_Y"])
# Specify the equivalent of "X JOIN Y on X.a_in_X = Y.b_in_Y"
# in a SQL SELECT statement.
df_joined = dfX.join(dfY, col("a_in_X") == col("b_in_Y")).select(dfX["a_in_X"].alias("the_joined_column"))
df_joined.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined

-----------------------
|"THE_JOINED_COLUMN"  |
-----------------------
|1                    |
-----------------------

Lorsque vous faites référence à des colonnes dans deux objets DataFrame différents qui ont le même nom (par exemple, en joignant les DataFrames sur cette colonne), vous pouvez utiliser la méthode DataFrame.col dans un objet DataFrame pour faire référence à une colonne dans cet objet (par exemple, df1.col("name") et df2.col("name")).

L’exemple suivant montre comment utiliser la méthode DataFrame.col pour faire référence à une colonne dans un DataFrame spécifique. L’exemple joint deux objets DataFrame qui ont tous deux une colonne nommée key. L’exemple utilise la méthode Column.as pour modifier les noms des colonnes dans le DataFrame nouvellement créé.

# Create two DataFrames to join
df_lhs = session.create_dataframe([["a", 1], ["b", 2]], schema=["key", "value"])
df_rhs = session.create_dataframe([["a", 3], ["b", 4]], schema=["key", "value"])
# Create a DataFrame that joins two other DataFrames (df_lhs and df_rhs).
# Use the DataFrame.col method to refer to the columns used in the join.
df_joined = df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).select(df_lhs.col("key").as_("key"), df_lhs.col("value").as_("L"), df_rhs.col("value").as_("R"))
df_joined.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_joined

---------------------
|"KEY"  |"L"  |"R"  |
---------------------
|a      |1    |3    |
|b      |2    |4    |
---------------------

Utilisation de guillemets doubles autour des identificateurs d’objets (noms de tables, noms de colonnes, etc.)¶

Les noms des bases de données, des schémas, des tables et des zones de préparation que vous spécifiez doivent être conformes aux exigences de l’identificateur Snowflake.

Créez une table dont les colonnes sont sensibles à la casse :

session.sql("""
  create or replace temp table "10tablename"(
  id123 varchar, -- case insensitive because it's not quoted.
  "3rdID" varchar, -- case sensitive.
  "id with space" varchar -- case sensitive.
)""").collect()
# Add return to the statement to return the collect() results in a Python worksheet

[Row(status='Table 10tablename successfully created.')]

Ajoutez ensuite des valeurs à la table :

session.sql("""insert into "10tablename" (id123, "3rdID", "id with space") values ('a', 'b', 'c')""").collect()
# Add return to the statement to return the collect() results in a Python worksheet

[Row(number of rows inserted=1)]

Créez ensuite un DataFrame pour la table et interrogez la table :

df = session.table('"10tablename"')
df.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df

---------------------------------------
|"ID123"  |"3rdID"  |"id with space"  |
---------------------------------------
|a        |b        |c                |
---------------------------------------

Lorsque vous spécifiez un nom, Snowflake considère que le nom est en majuscules. Par exemple, les appels suivants sont équivalents :

df.select(col("id123")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(ID123='a')]

Si le nom n’est pas conforme aux exigences de l’identificateur, vous devez utiliser des guillemets doubles (") autour du nom. Utilisez une barre oblique inverse (\) pour échapper le caractère de guillemet double dans un littéral de chaîne. Par exemple, le nom de la table suivante ne commence pas par une lettre ou un trait de soulignement, vous devez donc utiliser des guillemets autour du nom :

df = session.table("\"10tablename\"")

Vous pouvez également utiliser des guillemets simples à la place des barres obliques pour échapper au caractère guillemet double dans un littéral de chaîne.

df = session.table('"10tablename"')

Notez que lorsque vous spécifiez le nom d’une colonne, vous n’avez pas besoin d’utiliser des guillemets doubles autour du nom. La bibliothèque Snowpark met automatiquement le nom de la colonne entre guillemets doubles pour vous si le nom ne respecte pas les exigences d’identification :

df.select(col("3rdID")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(3rdID='b')]

Autre exemple, les appels suivants sont équivalents :

df.select(col("id with space")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(id with space='c')]

df.select(col("\"id with space\"")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(id with space='c')]

Si vous avez déjà ajouté des guillemets autour d’un nom de colonne, la bibliothèque n’insère pas de guillemets doubles supplémentaires autour du nom.

Dans certains cas, le nom de la colonne peut contenir des caractères guillemets doubles :

session.sql('''
  create or replace temp table quoted(
  "name_with_""air""_quotes" varchar,
  """column_name_quoted""" varchar
)''').collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(status='Table QUOTED successfully created.')]

session.sql('''insert into quoted ("name_with_""air""_quotes", """column_name_quoted""") values ('a', 'b')''').collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(number of rows inserted=1)]

Comme expliqué dans Exigences relatives à l’identificateur, pour chaque caractère entre guillemets doubles dans un identificateur entre guillemets doubles, vous devez utiliser deux caractères de guillemets doubles (par exemple, "name_with_""air""_quotes" et """column_name_quoted""") :

df_table = session.table("quoted")
df_table.select("\"name_with_\"\"air\"\"_quotes\"").collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(name_with_"air"_quotes='a')]

df_table.select("\"\"\"column_name_quoted\"\"\"").collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row("column_name_quoted"='b')]

Lorsqu’un identificateur est délimité par des guillemets doubles (que vous ayez explicitement ajouté des guillemets ou que la bibliothèque les ait ajoutés pour vous), Snowflake traite l’identificateur comme sensible à la casse :

# The following calls are NOT equivalent!
# The Snowpark library adds double quotes around the column name,
# which makes Snowflake treat the column name as case-sensitive.
df.select(col("id with space")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(id with space='c')]

En comparaison avec cet exemple :

from snowflake.snowpark.exceptions import SnowparkSQLException
try:
  df.select(col("ID WITH SPACE")).collect()
except SnowparkSQLException as e:
  print(e.message)

000904 (42000): SQL compilation error: error line 1 at position 7
invalid identifier '"ID WITH SPACE"'

Utilisation de littéraux comme objets de colonne¶

Pour utiliser un littéral dans une méthode qui prend un objet Column comme argument, créez un objet Column pour le littéral en transmettant le littéral à la fonction lit dans l’objet snowflake.snowpark.functions. Par exemple :

# Import for the lit and col functions.
from snowflake.snowpark.functions import col, lit

# Show the first 10 rows in which num_items is greater than 5.
# Use `lit(5)` to create a Column object for the literal 5.
df_filtered = df.filter(col("num_items") > lit(5))

Conversion d’un objet de colonne en un type spécifique¶

Pour convertir un objet Column en un type spécifique, appelez la méthode cast et transmettez un objet de type à partir du module snowflake.snowpark.types. Par exemple, pour convertir un littéral en NUMBER avec une précision de 5 et une échelle de 2 :

# Import for the lit function.
from snowflake.snowpark.functions import lit

 # Import for the DecimalType class.
from snowflake.snowpark.types import DecimalType

decimal_value = lit(0.05).cast(DecimalType(5,2))

Chaîner les appels de méthode¶

Parce que chaque méthode qui transforme un objet DataFrame renvoie un nouvel objet DataFrame auquel la transformation a été appliquée, vous pouvez chaîner les appels de méthode pour produire un nouvel objet DataFrame qui est transformé de manière supplémentaire.

L’exemple suivant renvoie un DataFrame qui est configuré pour :

Interroger la table sample_product_data.
Retourner la ligne avec id = 1.

Sélectionner les colonnes name et serial_number.

df_product_info = session.table("sample_product_data").filter(col("id") == 1).select(col("name"), col("serial_number"))
df_product_info.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_product_info

-------------------------------
|"NAME"     |"SERIAL_NUMBER"  |
-------------------------------
|Product 1  |prod-1           |
-------------------------------

Dans cet exemple :

session.table("sample_product_data") retourne un DataFrame pour la table sample_product_data.

Bien que le DataFrame ne contienne pas encore les données de la table, l’objet contient les définitions des colonnes de la table.
filter(col("id") == 1) retourne un DataFrame pour la table sample_product_data qui est configurée pour retourner la ligne avec id = 1.

Notez que le DataFrame ne contient pas encore la ligne correspondante de la table. La ligne correspondante n’est pas récupérée avant que vous appeliez une méthode d’action.
select(col("name"), col("serial_number")) renvoie un DataFrame qui contient les colonnes name et serial_number de la ligne de la table sample_product_data qui a id = 1.

L’ordre des appels est important lorsque vous enchaînez des appels de méthodes. Chaque appel de méthode renvoie un DataFrame qui a été transformé. Assurez-vous que les appels ultérieurs fonctionnent avec le DataFrame transformé.

Lorsque vous utilisez Snowpark Python, vous pouvez avoir besoin d’effectuer les appels de méthode select et filter dans un ordre différent de celui dans lequel vous utiliseriez les mots-clés équivalents (SELECT et WHERE) dans une instruction SQL.

Récupération des définitions de colonnes¶

Pour récupérer la définition des colonnes dans le jeu de données pour le DataFrame, appelez la propriété schema. Cette méthode renvoie un objet StructType qui contient un list d’objets StructField. Chaque objet StructField contient la définition d’une colonne.

# Import the StructType
from snowflake.snowpark.types import *
# Get the StructType object that describes the columns in the
# underlying rowset.
table_schema = session.table("sample_product_data").schema
table_schema
StructType([StructField('ID', LongType(), nullable=True), StructField('PARENT_ID', LongType(), nullable=True), StructField('CATEGORY_ID', LongType(), nullable=True), StructField('NAME', StringType(), nullable=True), StructField('SERIAL_NUMBER', StringType(), nullable=True), StructField('KEY', LongType(), nullable=True), StructField('"3rd"', LongType(), nullable=True)])

Dans l’objet StructType retourné, les noms de colonnes sont toujours normalisés. Les identificateurs non mis entre guillemets simples sont renvoyés en majuscules, et les identificateurs mis entre guillemets simples sont renvoyés dans la casse exacte dans laquelle ils ont été définis.

L’exemple suivant crée un DataFrame contenant les colonnes nommées ID et 3rd. Pour le nom de colonne 3rd, la bibliothèque Snowpark place automatiquement le nom entre guillemets doubles ("3rd") car le nom ne répond pas aux exigences d’un identificateur.

L’exemple appelle la propriété schema puis la propriété names sur l’objet StructType renvoyé pour obtenir un list de noms de colonnes. Les noms sont normalisés dans les StructType renvoyés par la propriété schema.

# Create a DataFrame containing the "id" and "3rd" columns.
df_selected_columns = session.table("sample_product_data").select(col("id"), col("3rd"))
# Print out the names of the columns in the schema.
# This prints List["ID", "\"3rd\""]
df_selected_columns.schema.names

['ID', '"3rd"']

Exécution d’une action pour évaluer un DataFrame¶

Comme nous l’avons mentionné précédemment, le DataFrame est évalué de façon « lazy », ce qui signifie que l’instruction SQL n’est pas envoyée au serveur pour être exécutée tant que vous n’avez pas effectué une action. Une action provoque l’évaluation du DataFrame et envoie l’instruction SQL correspondante au serveur pour exécution.

Les méthodes suivantes exécutent une action :


Classe	Méthode	Description
`DataFrame`	`collect`	Évalue les DataFrame et renvoie le jeu de données résultant sous la forme d’un `list` d’objets `Row`.
`DataFrame`	`count`	Évalue le DataFrame et retourne le nombre de lignes.
`DataFrame`	`show`	Évalue le DataFrame et imprime les lignes dans la console. Cette méthode limite le nombre de lignes à 10 (par défaut).
`DataFrameWriter`	`save_as_table`	Sauvegarde les données du DataFrame dans la table spécifiée. Reportez-vous à Sauvegarde des données dans une table.

Par exemple, pour exécuter une requête sur une table et renvoyer les résultats, appelez la méthode collect :

# Create a DataFrame with the "id" and "name" columns from the "sample_product_data" table.
# This does not execute the query.
df = session.table("sample_product_data").select(col("id"), col("name"))

# Send the query to the server for execution and
# return a list of Rows containing the results.
results = df.collect()
# Use a return statement to return the collect() results in a Python worksheet
# return results

Pour exécuter la requête et renvoyer le nombre de résultats, appelez la méthode count :

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Send the query to the server for execution and
# print the count of rows in the table.
print(df_products.count())
12

Pour exécuter une requête et imprimer les résultats dans la console, appelez la méthode show :

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Send the query to the server for execution and
# print the results to the console.
# The query limits the number of rows to 10 by default.
df_products.show()
# To return the DataFrame as a table in a Python worksheet use return instead of show()
return df_products

-------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"      |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
-------------------------------------------------------------------------------------
|1     |0            |5              |Product 1   |prod-1           |1      |10     |
|2     |1            |5              |Product 1A  |prod-1-A         |1      |20     |
|3     |1            |5              |Product 1B  |prod-1-B         |1      |30     |
|4     |0            |10             |Product 2   |prod-2           |2      |40     |
|5     |4            |10             |Product 2A  |prod-2-A         |2      |50     |
|6     |4            |10             |Product 2B  |prod-2-B         |2      |60     |
|7     |0            |20             |Product 3   |prod-3           |3      |70     |
|8     |7            |20             |Product 3A  |prod-3-A         |3      |80     |
|9     |7            |20             |Product 3B  |prod-3-B         |3      |90     |
|10    |0            |50             |Product 4   |prod-4           |4      |100    |
-------------------------------------------------------------------------------------

Pour limiter le nombre de lignes à 20 :

# Create a DataFrame for the "sample_product_data" table.
df_products = session.table("sample_product_data")

# Limit the number of rows to 20, rather than 10.
df_products.show(20)
# All rows are returned when you use return in a Python worksheet to return the DataFrame as a table
return df_products

-------------------------------------------------------------------------------------
|"ID"  |"PARENT_ID"  |"CATEGORY_ID"  |"NAME"      |"SERIAL_NUMBER"  |"KEY"  |"3rd"  |
-------------------------------------------------------------------------------------
|1     |0            |5              |Product 1   |prod-1           |1      |10     |
|2     |1            |5              |Product 1A  |prod-1-A         |1      |20     |
|3     |1            |5              |Product 1B  |prod-1-B         |1      |30     |
|4     |0            |10             |Product 2   |prod-2           |2      |40     |
|5     |4            |10             |Product 2A  |prod-2-A         |2      |50     |
|6     |4            |10             |Product 2B  |prod-2-B         |2      |60     |
|7     |0            |20             |Product 3   |prod-3           |3      |70     |
|8     |7            |20             |Product 3A  |prod-3-A         |3      |80     |
|9     |7            |20             |Product 3B  |prod-3-B         |3      |90     |
|10    |0            |50             |Product 4   |prod-4           |4      |100    |
|11    |10           |50             |Product 4A  |prod-4-A         |4      |100    |
|12    |10           |50             |Product 4B  |prod-4-B         |4      |100    |
-------------------------------------------------------------------------------------

Note

Si vous appelez la propriété schema pour obtenir les définitions des colonnes dans le DataFrame, vous n’avez pas besoin d’appeler une méthode d’action.

Sauvegarde des données dans une table¶

Pour enregistrer le contenu d’un DataFrame dans une table :

Appelez la propriété write pour obtenir un objet DataFrameWriter.
Appelez la méthode mode dans l’objet DataFrameWriter et spécifiez le mode. Pour plus d’informations, consultez la documentation sur l’API. Cette méthode renvoie un nouvel objet DataFrameWriter qui est configuré avec le mode spécifié.
Appelez la méthode save_as_table dans l’objet DataFrameWriter pour sauvegarder le contenu du DataFrame dans une table spécifiée.

Notez que vous n’avez pas besoin d’appeler une méthode distincte (par exemple collect) pour exécuter l’instruction SQL qui enregistre les données dans la table.

Par exemple :

df.write.mode("overwrite").save_as_table("table1")

Création d’une vue à partir d’un DataFrame¶

Pour créer une vue à partir d’un DataFrame, appelez la méthode create_or_replace_view, qui crée immédiatement la nouvelle vue :

import os
database = os.environ["snowflake_database"]  # use your own database and schema
schema = os.environ["snowflake_schema"]
view_name = "my_view"
df.create_or_replace_view(f"{database}.{schema}.{view_name}")

[Row(status='View MY_VIEW successfully created.')]

Dans une feuille de calcul Python, étant donné que vous exécutez la feuille de calcul dans le contexte d’une base de données et d’un schéma, vous pouvez exécuter ce qui suit pour créer une vue :

# Define a DataFrame
df_products = session.table("sample_product_data")
# Define a View name
view_name = "my_view"
# Create the view
df_products.create_or_replace_view(f"{view_name}")
# return the view name
return view_name + " successfully created"
my_view successfully created

Les vues que vous créez en appelant create_or_replace_view sont persistantes. Si vous n’avez plus besoin de cette vue, vous pouvez détruire la vue manuellement.

Vous pouvez également utiliser la méthode create_or_replace_temp_view qui crée une vue temporaire. La vue temporaire n’est disponible que dans la session dans laquelle elle est créée.

Travailler avec des fichiers dans une zone de préparation¶

Cette section explique comment interroger les données d’un fichier dans une zone de préparation Snowflake. Pour les autres opérations sur les fichiers, utilisez des instructions SQL.

Pour interroger des données dans des fichiers dans une zone de préparation Snowflake, utilisez la classe DataFrameReader :

Appelez la méthode read de la classe Session pour accéder à un objet DataFrameReader.
Si les fichiers sont au format CSV, décrivez les champs du fichier. Pour ce faire :
1. Créez un objet StructType composé d’un list d’objets StructField qui décrivent les champs du fichier.
2. Pour chaque objet StructField indiquez ce qui suit :
  - Nom du champ.
  - Le type de données du champ (spécifié comme un objet dans le module snowflake.snowpark.types).
  - Indique si le champ peut être « Null » ou non.
  Par exemple :
  from snowflake.snowpark.types import * schema_for_data_file = StructType([ StructField("id", StringType()), StructField("name", StringType()) ])
3. Appelez la propriété schema dans l’objet DataFrameReader en transmettant l’objet StructType.
  
  Par exemple :
  df_reader = session.read.schema(schema_for_data_file)
  La propriété schema renvoie un objet DataFrameReader qui est configuré pour lire les fichiers contenant les champs spécifiés.
  
  Notez que vous n’avez pas besoin de faire cela pour les fichiers dans d’autres formats (tels que JSON). Pour ces fichiers, le DataFrameReader traite les données comme un champ unique de type VARIANT avec le nom de champ $1.
Si vous devez spécifier des informations supplémentaires sur la manière dont les données doivent être lues (par exemple, que les données sont compressées ou qu’un fichier CSV utilise un point-virgule au lieu d’une virgule pour délimiter les champs), appelez les méthodes option ou options de l’objet DataFrameReader.

La méthode option prend un nom et une valeur de l’option que vous voulez définir et vous permet de combiner plusieurs appels chaînés, lorsque la méthode options prend un dictionnaire des noms des options et de leurs valeurs correspondantes.

Pour connaître les noms et les valeurs des options de format de fichier, consultez la documentation sur CREATE FILE FORMAT.

Vous pouvez également définir les options de copie décrites dans la documentation COPY INTO TABLE. Notez que la définition des options de copie peut entraîner une stratégie d’exécution plus coûteuse lorsque vous récupérez les données dans le DataFrame.

L’exemple suivant configure l’objet DataFrameReader pour interroger les données d’un fichier CSV qui n’est pas compressé et qui utilise un point-virgule comme délimiteur de champ.
df_reader = df_reader.option("field_delimiter", ";").option("COMPRESSION", "NONE")
Les méthodes option et options renvoient un objet DataFrameReader qui est configuré avec les options spécifiées.
Appelez la méthode correspondant au format du fichier (par exemple, la méthode csv), en transmettant l’emplacement du fichier.
df = df_reader.csv("@s3_ts_stage/emails/data_0_0_0.csv")
Les méthodes correspondant au format d’un fichier renvoient un objet DataFrame qui est configuré pour contenir les données de ce fichier.
Utilisez les méthodes de l’objet DataFrame pour effectuer toute transformation nécessaire sur l’ensemble de données (par exemple, sélection de champs spécifiques, filtrage des lignes, etc.)

Par exemple, pour extraire l’élément color d’un fichier JSON dans la zone de préparation nommée my_stage :
# Import the sql_expr function from the functions module. from snowflake.snowpark.functions import sql_expr df = session.read.json("@my_stage").select(sql_expr("$1:color"))
Comme expliqué précédemment, pour les fichiers dans des formats autres que CSV (par exemple JSON), le DataFrameReader traite les données du fichier comme une seule colonne VARIANT portant le nom $1.

Cet exemple utilise la fonction sql_expr dans le module snowflake.snowpark.functions pour spécifier le chemin à l’élément color.

Notez que la fonction sql_expr n’interprète ni ne modifie l’argument d’entrée. La fonction vous permet simplement de construire des expressions et des snippets dans SQL qui ne sont pas encore pris en charge par l’API Snowpark.
Appelez une méthode d’action pour interroger les données dans le fichier.

Comme c’est le cas avec les DataFrames pour les tables, les données ne sont pas récupérées dans les DataFrame avant que vous n’appeliez une méthode d’action.

Utilisation de données semi-structurées¶

En utilisant un DataFrame, vous pouvez interroger et accéder à des données semi-structurées (par exemple, des données JSON). Les sections suivantes expliquent comment travailler avec des données semi-structurées dans un DataFrame.

Parcours de données semi-structurées
Conversion explicite des valeurs en données semi-structurées
Aplatissement d’un tableau d’objets en lignes

Note

Les exemples de ces sections utilisent les données d’exemple dans Échantillon de données utilisé dans des exemples.

Parcours de données semi-structurées¶

Pour faire référence à un champ ou à un élément spécifique dans des données semi-structurées, utilisez les méthodes suivantes de l’objet Column :

Obtenez l’attribut col_object["<field_name>"] pour retourner un objet Column pour un champ dans un OBJECT (ou un VARIANT qui contient un OBJECT).
Utilisez col_object[<index>] pour renvoyer un objet Column pour un élément dans un ARRAY (ou un VARIANT qui contient un ARRAY).

Note

Si le nom du champ ou les éléments du chemin sont irréguliers et rendent difficile l’utilisation de l’indexation décrite ci-dessous, vous pouvez utiliser get, get_ignore_case, ou get_path comme alternative.

Par exemple, le code suivant sélectionne le champ dealership dans les objets de la colonne src des données d’exemple :

from snowflake.snowpark.functions import col

df = session.table("car_sales")
df.select(col("src")["dealership"]).show()

Le code imprime la sortie suivante :

----------------------------
|"""SRC""['DEALERSHIP']"   |
----------------------------
|"Valley View Auto Sales"  |
|"Tindel Toyota"           |
----------------------------

Note

Les valeurs contenues dans le DataFrame sont délimitées par des guillemets doubles car elles sont renvoyées sous forme de littéraux de chaînes. Pour convertir ces valeurs en un type spécifique, voir Conversion explicite des valeurs en données semi-structurées.

Vous pouvez également enchaîner les appels de méthode pour parcourir un chemin vers un champ ou un élément spécifique.

Par exemple, le code suivant sélectionne le champ name dans l’objet salesperson :

df = session.table("car_sales")
df.select(df["src"]["salesperson"]["name"]).show()

Le code imprime la sortie suivante :

------------------------------------
|"""SRC""['SALESPERSON']['NAME']"  |
------------------------------------
|"Frank Beasley"                   |
|"Greg Northrup"                   |
------------------------------------

Autre exemple, le code suivant sélectionne le premier élément du champ vehicle, qui contient un tableau de véhicules. L’exemple sélectionne également le champ price du premier élément.

df = session.table("car_sales")
df.select(df["src"]["vehicle"][0]).show()
df.select(df["src"]["vehicle"][0]["price"]).show()

Le code imprime la sortie suivante :

---------------------------
|"""SRC""['VEHICLE'][0]"  |
---------------------------
|{                        |
|  "extras": [            |
|    "ext warranty",      |
|    "paint protection"   |
|  ],                     |
|  "make": "Honda",       |
|  "model": "Civic",      |
|  "price": "20275",      |
|  "year": "2017"         |
|}                        |
|{                        |
|  "extras": [            |
|    "ext warranty",      |
|    "rust proofing",     |
|    "fabric protection"  |
|  ],                     |
|  "make": "Toyota",      |
|  "model": "Camry",      |
|  "price": "23500",      |
|  "year": "2017"         |
|}                        |
---------------------------

------------------------------------
|"""SRC""['VEHICLE'][0]['PRICE']"  |
------------------------------------
|"20275"                           |
|"23500"                           |
------------------------------------

Au lieu d’accéder aux champs de la manière susmentionnée, vous pouvez utiliser les fonctions get, get_ignore_case, ou get_path si le nom du champ ou les éléments du chemin sont irréguliers.

Par exemple, les lignes de code suivantes impriment toutes deux la valeur d’un champ spécifié dans un objet :

from snowflake.snowpark.functions import get, get_path, lit

df.select(get(col("src"), lit("dealership"))).show()
df.select(col("src")["dealership"]).show()

De même, les lignes de code suivantes impriment toutes deux la valeur d’un champ à un chemin spécifié dans un objet :

df.select(get_path(col("src"), lit("vehicle[0].make"))).show()
df.select(col("src")["vehicle"][0]["make"]).show()

Conversion explicite des valeurs en données semi-structurées¶

Par défaut, les valeurs des champs et des éléments sont renvoyées sous forme de littéraux de chaînes (y compris les guillemets), comme le montrent les exemples ci-dessus.

Pour éviter les résultats inattendus, appelez la méthode cast pour convertir la valeur en un type spécifique. Par exemple, le code suivant imprime les valeurs sans et avec conversion :

# Import the objects for the data types, including StringType.
from snowflake.snowpark.types import *

df = session.table("car_sales")
df.select(col("src")["salesperson"]["id"]).show()
df.select(col("src")["salesperson"]["id"].cast(StringType())).show()

Le code imprime la sortie suivante :

----------------------------------
|"""SRC""['SALESPERSON']['ID']"  |
----------------------------------
|"55"                            |
|"274"                           |
----------------------------------

---------------------------------------------------
|"CAST (""SRC""['SALESPERSON']['ID'] AS STRING)"  |
---------------------------------------------------
|55                                               |
|274                                              |
---------------------------------------------------

Aplatissement d’un tableau d’objets en lignes¶

Si vous devez « aplatir » des données semi-structurées dans un DataFrame (par exemple, produire une ligne pour chaque objet d’un tableau), appelez la fonction flatten en utilisant la méthode join_table_function. Cette méthode est équivalente à la fonction FLATTEN SQL. Si vous transmettez un chemin d’accès à un objet ou à un tableau, la méthode renvoie un DataFrame qui contient une ligne pour chaque champ ou élément de l’objet ou du tableau.

Par exemple, dans les données d’exemple, src:customer est un tableau d’objets qui contient des informations sur un client. Chaque objet contient un champ name et address.

Si vous transmettez ce chemin à la fonction flatten :

df = session.table("car_sales")
df.join_table_function("flatten", col("src")["customer"]).show()

cette méthode renvoie un DataFrame :

----------------------------------------------------------------------------------------------------------------------------------------------------------
|"SRC"                                      |"SEQ"  |"KEY"  |"PATH"  |"INDEX"  |"VALUE"                            |"THIS"                               |
----------------------------------------------------------------------------------------------------------------------------------------------------------
|{                                          |1      |NULL   |[0]     |0        |{                                  |[                                    |
|  "customer": [                            |       |       |        |         |  "address": "San Francisco, CA",  |  {                                  |
|    {                                      |       |       |        |         |  "name": "Joyce Ridgely",         |    "address": "San Francisco, CA",  |
|      "address": "San Francisco, CA",      |       |       |        |         |  "phone": "16504378889"           |    "name": "Joyce Ridgely",         |
|      "name": "Joyce Ridgely",             |       |       |        |         |}                                  |    "phone": "16504378889"           |
|      "phone": "16504378889"               |       |       |        |         |                                   |  }                                  |
|    }                                      |       |       |        |         |                                   |]                                    |
|  ],                                       |       |       |        |         |                                   |                                     |
|  "date": "2017-04-28",                    |       |       |        |         |                                   |                                     |
|  "dealership": "Valley View Auto Sales",  |       |       |        |         |                                   |                                     |
|  "salesperson": {                         |       |       |        |         |                                   |                                     |
|    "id": "55",                            |       |       |        |         |                                   |                                     |
|    "name": "Frank Beasley"                |       |       |        |         |                                   |                                     |
|  },                                       |       |       |        |         |                                   |                                     |
|  "vehicle": [                             |       |       |        |         |                                   |                                     |
|    {                                      |       |       |        |         |                                   |                                     |
|      "extras": [                          |       |       |        |         |                                   |                                     |
|        "ext warranty",                    |       |       |        |         |                                   |                                     |
|        "paint protection"                 |       |       |        |         |                                   |                                     |
|      ],                                   |       |       |        |         |                                   |                                     |
|      "make": "Honda",                     |       |       |        |         |                                   |                                     |
|      "model": "Civic",                    |       |       |        |         |                                   |                                     |
|      "price": "20275",                    |       |       |        |         |                                   |                                     |
|      "year": "2017"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
|{                                          |2      |NULL   |[0]     |0        |{                                  |[                                    |
|  "customer": [                            |       |       |        |         |  "address": "New York, NY",       |  {                                  |
|    {                                      |       |       |        |         |  "name": "Bradley Greenbloom",    |    "address": "New York, NY",       |
|      "address": "New York, NY",           |       |       |        |         |  "phone": "12127593751"           |    "name": "Bradley Greenbloom",    |
|      "name": "Bradley Greenbloom",        |       |       |        |         |}                                  |    "phone": "12127593751"           |
|      "phone": "12127593751"               |       |       |        |         |                                   |  }                                  |
|    }                                      |       |       |        |         |                                   |]                                    |
|  ],                                       |       |       |        |         |                                   |                                     |
|  "date": "2017-04-28",                    |       |       |        |         |                                   |                                     |
|  "dealership": "Tindel Toyota",           |       |       |        |         |                                   |                                     |
|  "salesperson": {                         |       |       |        |         |                                   |                                     |
|    "id": "274",                           |       |       |        |         |                                   |                                     |
|    "name": "Greg Northrup"                |       |       |        |         |                                   |                                     |
|  },                                       |       |       |        |         |                                   |                                     |
|  "vehicle": [                             |       |       |        |         |                                   |                                     |
|    {                                      |       |       |        |         |                                   |                                     |
|      "extras": [                          |       |       |        |         |                                   |                                     |
|        "ext warranty",                    |       |       |        |         |                                   |                                     |
|        "rust proofing",                   |       |       |        |         |                                   |                                     |
|        "fabric protection"                |       |       |        |         |                                   |                                     |
|      ],                                   |       |       |        |         |                                   |                                     |
|      "make": "Toyota",                    |       |       |        |         |                                   |                                     |
|      "model": "Camry",                    |       |       |        |         |                                   |                                     |
|      "price": "23500",                    |       |       |        |         |                                   |                                     |
|      "year": "2017"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
----------------------------------------------------------------------------------------------------------------------------------------------------------

À partir de ce DataFrame, vous pouvez sélectionner les champs name et address de chaque objet dans le champ VALUE :

df.join_table_function("flatten", col("src")["customer"]).select(col("value")["name"], col("value")["address"]).show()

-------------------------------------------------
|"""VALUE""['NAME']"   |"""VALUE""['ADDRESS']"  |
-------------------------------------------------
|"Joyce Ridgely"       |"San Francisco, CA"     |
|"Bradley Greenbloom"  |"New York, NY"          |
-------------------------------------------------

Le code suivant complète l’exemple précédent en convertissant les valeurs en un type spécifique et en modifiant les noms des colonnes :

df.join_table_function("flatten", col("src")["customer"]).select(col("value")["name"].cast(StringType()).as_("Customer Name"), col("value")["address"].cast(StringType()).as_("Customer Address")).show()

-------------------------------------------
|"Customer Name"     |"Customer Address"  |
-------------------------------------------
|Joyce Ridgely       |San Francisco, CA   |
|Bradley Greenbloom  |New York, NY        |
-------------------------------------------

Exécution d’instructions SQL¶

Pour exécuter une instruction SQL que vous spécifiez, appelez la méthode sql de la classe Session, et transmettez l’instruction à exécuter. Cette méthode renvoie un DataFrame.

Notez que l’instruction SQL ne sera pas exécutée tant que vous n’aurez pas appelé une méthode d’action.

# Get the list of the files in a stage.
# The collect() method causes this SQL statement to be executed.
session.sql("create or replace temp stage my_stage").collect()

# Prepend a return statement to return the collect() results in a Python worksheet
[Row(status='Stage area MY_STAGE successfully created.')]

stage_files_df = session.sql("ls @my_stage").collect()
# Prepend a return statement to return the collect() results in a Python worksheet
# Resume the operation of a warehouse.
# Note that you must call the collect method to execute
# the SQL statement.
session.sql("alter warehouse if exists my_warehouse resume if suspended").collect()

# Prepend a return statement to return the collect() results in a Python worksheet
[Row(status='Statement executed successfully.')]

# Set up a SQL statement to copy data from a stage to a table.
session.sql("copy into sample_product_data from @my_stage file_format=(type = csv)").collect()
# Prepend a return statement to return the collect() results in a Python worksheet

[Row(status='Copy executed with 0 files processed.')]

Si vous voulez appeler des méthodes pour transformer le DataFrame (par exemple, filter, select, etc.), notez que ces méthodes ne fonctionnent que si l’instruction SQL sous-jacente est une instruction SELECT. Les méthodes de transformation ne sont pas prises en charge pour les autres types d’instructions SQL.

df = session.sql("select id, parent_id from sample_product_data where id < 10")
# Because the underlying SQL statement for the DataFrame is a SELECT statement,
# you can call the filter method to transform this DataFrame.
results = df.filter(col("id") < 3).select(col("id")).collect()
# Prepend a return statement to return the collect() results in a Python worksheet

# In this example, the underlying SQL statement is not a SELECT statement.
df = session.sql("ls @my_stage")
# Calling the filter method results in an error.
try:
  df.filter(col("size") > 50).collect()
except SnowparkSQLException as e:
  print(e.message)

000904 (42000): SQL compilation error: error line 1 at position 104
invalid identifier 'SIZE'

Soumettre simultanément des requêtes Snowpark¶

Note

Cette fonction requiert la bibliothèque Snowpark pour Python version 1.24 ou supérieure et la version 8.46 ou supérieure du serveur.

Les objets de session « Thread-safe » permettent à différentes parties de votre code Python Snowpark de s’exécuter simultanément tout en utilisant la même session. Cela permet d’exécuter simultanément plusieurs opérations - telles que des transformations sur plusieurs DataFrames. Ceci est particulièrement utile lorsque vous travaillez avec des requêtes qui peuvent être traitées indépendamment sur le serveur Snowflake et cela s’aligne sur une approche multithreading plus traditionnelle.

Le Global Interpreter Lock (GIL) en Python est un mutex qui protège l’accès aux objets Python, empêchant plusieurs threads natifs d’exécuter simultanément le bytecode Python. Alors que les opérations liées aux E/S peuvent toujours bénéficier du modèle de threading de Python en raison de la version GIL libérée pendant les opérations d’E/S, les threads liés au CPU n’atteindront pas un véritable parallélisme car un seul thread peut s’exécuter à la fois.

De plus, lorsqu’il est utilisé à l’intérieur de Snowflake (par exemple dans une procédure stockée), le serveur Python Snowpark gère le Global Interpreter Lock (GIL) en le libérant avant de soumettre des requêtes à Snowflake. Cela permet de garantir une véritable concurrence lorsque plusieurs requêtes sont mises en file d’attente à partir de threads distincts. Grâce à cette gestion, Snowpark permet à plusieurs threads de soumettre des requêtes simultanément, ce qui garantit une exécution parallèle optimale.

Avantages de l’utilisation d’objets de session « Thread-safe » dans Snowpark¶

La possibilité d’effectuer plusieurs opérations DataFrame simultanément peut apporter les avantages suivants aux utilisateurs de Snowpark :

Amélioration des performances : les objets de session « Thread-safe » vous permettent d’exécuter plusieurs requêtes Snowpark Python simultanément, réduisant ainsi l’environnement d’exécution global. Par exemple, si vous devez traiter plusieurs tables indépendamment les unes des autres, cette fonction permet de réduire considérablement le temps nécessaire à l’exécution du travail, car vous n’avez plus besoin d’attendre la fin du traitement de chaque table avant de commencer le suivant.
Utilisation efficace des ressources informatiques : l’envoi simultané de requêtes garantit une utilisation efficace des ressources de calcul de Snowflake, réduisant ainsi les temps morts.
Facilité d’utilisation : les objets de session « Thread-safe » s’intègrent de manière transparente aux APIs multithreading natives de Python, ce qui permet aux développeurs d’exploiter les outils intégrés de Python pour contrôler le comportement des threads et optimiser l’exécution parallèle.

Les objets de session « Thread-safe » et les tâches asynchrones peuvent se compléter en fonction de votre cas d’utilisation. Les travaux asynchrones sont utiles lorsque vous n’avez pas besoin d’attendre que vos travaux se terminent, ce qui permet une exécution non bloquante sans gestion du pool de threads. Les objets de session « Thread-safe », en revanche, sont utiles lorsque vous souhaitez soumettre plusieurs requêtes simultanément du côté client. Dans certains cas, les blocs de code peuvent également contenir des tâches asynchrones, ce qui permet d’utiliser efficacement les deux méthodes ensemble.

Voici quelques exemples où les objets de session « Thread-safe » peuvent améliorer votre pipeline de données.

Exemple 1 : Chargement simultané de plusieurs tables¶

Cet exemple montre comment charger les données de trois fichiers CSV différents dans trois tables distinctes en utilisant trois threads pour exécuter la commande COPY INTO simultanément.

import threading
from snowflake.snowpark import Session

# Define the list of tables
tables = ["customers", "orders", "products"]

# Function to copy data from stage to tables
def execute_copy(table_name):
    try:
        # Read data from the stage using DataFrameReader
        df = (
            session.read.option("SKIP_HEADER", 1)
            .option("PATTERN", f"{table_name}[.]csv")
            .option("FORCE", True)
            .csv(f"@my_stage")
        )

        # Copy data into the target table
        df.copy_into_table(
            table_name=table_name, target_columns=session.table(table_name).columns
        )

    except Exception as e:
        print(f"Failed to copy data into {table_name}, Error: {e}")

# Create an empty list of threads
threads = []

# Loop through and start a thread for each table
for table in tables:
    thread = threading.Thread(target=execute_copy, args=(table,))
    threads.append(thread)
    thread.start()

# Wait for all threads to finish
for thread in threads:
    thread.join()

Exemple 2 : Traitement simultané de plusieurs tables¶

Cet exemple montre comment vous pouvez utiliser plusieurs threads pour filtrer, agréger et insérer simultanément des données dans une table de résultats à partir de chaque table de transaction client (transaction_client1, transaction_client2 et transaction_client3).

from concurrent.futures import ThreadPoolExecutor
from snowflake.snowpark import Session
from snowflake.snowpark.functions import col, month, sum, lit

# List of customers
customers = ["customer1", "customer2", "customer3"]

# Define a function to process each customer transaction table
def process_customer_table(customer_name):
    table_name = f"transaction_{customer_name}"

    try:
        # Load the customer transaction table
        df = session.table(table_name)
        print(f"Processing {table_name}...")

        # Filter data by positive values and non null categories
        df_filtered = df.filter((col("value") > 0) & col("category").is_not_null())

        # Perform aggregation: Sum of value by category and month
        df_aggregated = df_filtered.with_column("month", month(col("date"))).with_column("customer_name", lit(customer_name)).group_by(col("category"), col("month"), col("customer_name")).agg(sum("value").alias("total_value"))

        # Save the processed data into a new result table
        df_aggregated.show()
        df_aggregated.write.save_as_table("aggregate_customers", mode="append")
        print(f"Data from {table_name} processed and saved")

    except Exception as e:
        print(f"Error processing {table_name}: {e}")

# Using ThreadPoolExecutor to handle concurrency
with ThreadPoolExecutor(max_workers=3) as executor:
    # Submit tasks for each customer table
    executor.map(process_customer_table, customers)

# Display the results from the aggregate table
session.table("aggregate_customers").show()

Limites de l’utilisation d’objets de session « Thread-safe »¶

Si vous devez gérer plusieurs transactions simultanément, il est important d’utiliser plusieurs objets de session, car plusieurs threads d’une même session ne prennent pas en charge les transactions simultanées.
La modification des configurations d’exécution de la session (y compris les variables de session de Snowflake comme la base de données, le schéma, l’entrepôt, et les configurations côté client comme cte_optimization_enabled, sql_simplifier_enabled) alors que d’autres threads sont actifs peut conduire à un comportement inattendu. Pour éviter les conflits, il est préférable d’utiliser des objets de session distincts si différentes threads nécessitent des configurations distinctes. Par exemple, si vous devez effectuer des opérations sur différentes bases de données en parallèle, assurez-vous que chaque thread dispose de son propre objet de session plutôt que de partager la même session.

Renvoyer le contenu d’un DataFrame sous forme de DataFrame Pandas¶

Pour renvoyer le contenu d’un DataFrame sous la forme d’un DataFrame Pandas, utilisez la méthode to_pandas.

Par exemple :

python_df = session.create_dataframe(["a", "b", "c"])
pandas_df = python_df.to_pandas()

DataFrames Snowpark contre DataFrame pandas Snowpark : lequel dois-je choisir ?¶

En installant la bibliothèque Python Snowpark, vous avez la possibilité d’utiliser l”API DataFrames ou pandas on Snowflake.

Les DataFrames Snowpark sont modélisés à partir de PySpark, tandis que Snowpark pandas vise à étendre les fonctionnalités de Snowpark DataFrame et à fournir une interface familière aux utilisateurs de pandas afin de faciliter la migration et l’adoption. Nous vous recommandons d’utiliser les différentes APIs en fonction de votre cas d’utilisation et de vos préférences :


Utilisez Snowpark pandas si vous….	Utiliser des DataFrames Snowpark si vous…
préférez travailler avec ou avoir du code existant écrit en pandas	préférez travailler avec ou avoir du code existant écrit dans Spark
avez un flux de travail qui implique une analyse interactive et une exploration itérative	avez un flux de travail qui implique un traitement par lots et un développement itératif limité
êtes familier avec le travail avec des opérations de DataFrame qui sont exécutées immédiatement	êtes familier avec le travail avec des opérations de DataFrame qui sont évaluées paresseusement
préférez que les données soient cohérentes et ordonnées pendant les opérations	êtes d’accord avec le fait que les données ne soient pas ordonnées
êtes d’accord avec des performances légèrement plus lentes par rapport aux DataFrames Snowpark en faveur d’une utilisation plus facile des API	la performance est plus importante pour vous que la facilité d’utilisation

Du point de vue de la mise en œuvre, les DataFrames Snowpark et les DataFrames pandas sont sémantiquement différents. Puisque les DataFrames Snowpark sont modélisés d’après PySpark, ils opèrent sur la source de données originale, obtiennent les données mises à jour les plus récentes et ne maintiennent pas l’ordre des opérations. Les Snowpark pandas sont modélisés d’après pandas, et opèrent sur un instantané des données, maintiennent l’ordre pendant l’opération et permettent une indexation positionnelle basée sur l’ordre. Le maintien de l’ordre est utile pour l’inspection visuelle des données dans l’analyse interactive des données.

Pour plus d’informations, voir Utilisation de pandas on Snowflake avec les DataFrames Snowpark.