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.

Dans ce chapitre :

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 :

  1. 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.

  2. 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.

  3. 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 :

>>> 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)]

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 :

  • 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 from specified values.
>>> df1 = session.create_dataframe([1, 2, 3, 4]).to_df("a")  # one column, named "a"
>>> df1.show()
-------
|"A"  |
-------
|1    |
|2    |
|3    |
|4    |
-------


>>> # 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()
-------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------


>>> # 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()
-------------------------
|"A"  |"B"  |"C"  |"D"  |
-------------------------
|1    |2    |3    |4    |
-------------------------


>>> # 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()
---------------
|"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 will contain rows with values 1, 3, 5, 7, and 9 respectively.
>>> df_range = session.range(1, 10, 2).to_df("a")
>>> df_range.show()
-------
|"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()
--------------
|"NAME"      |
--------------
|Product 1   |
|Product 1A  |
|Product 1B  |
|Product 2   |
|Product 2A  |
|Product 2B  |
|Product 3   |
|Product 3A  |
|Product 3B  |
|Product 4   |
--------------

    Bien que vous puissiez utiliser cette méthode pour exécuter des instructions SELECT qui récupèrent des données dans des tables et des fichiers en zone de préparation, vous devriez plutôt utiliser la méthode table et la propriété read , qui peuvent fournir une meilleure coloration syntaxique, une meilleure mise en évidence des erreurs et une meilleure complétion de code intelligente dans les outils de développement.

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

Pour spécifier quelles colonnes doivent être sélectionnées et comment les résultats doivent être filtrés, triés, groupés, etc., 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. (Voir 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.
>>> 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()
------------------------------------------------------------------------------------
|"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()
---------------------------------------
|"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 original). Cela signifie que 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.

Notez que ces méthodes de transformation 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). Les méthodes de transformation spécifient simplement comment l’instruction SQL doit être construite.

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()
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------


>>> # Both dataframes have the same column "key", the following is more convenient.
>>> df_lhs.join(df_rhs, ["key"]).show()
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------


>>> # 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()
-------------------------------
|"KEY"  |"VALUE1"  |"VALUE2"  |
-------------------------------
|a      |1         |3         |
|b      |2         |4         |
-------------------------------


>>> # copy the DataFrame if you want to do a self-join
>>> from copy import copy
>>> 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")))

Notez que lorsqu’il y a des colonnes qui se chevauchent dans les Dataframes, Snowpark ajoutera un préfixe généré aléatoirement aux colonnes dans le résultat de la jointure :

>>> df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key")).show()  
-----------------------------------------------------
|"l_av5t_KEY"  |"VALUE1"  |"r_1p6k_KEY"  |"VALUE2"  |
-----------------------------------------------------
|a             |1         |a             |3         |
|b             |2         |b             |4         |
-----------------------------------------------------

Vous pouvez référencer les colonnes qui se chevauchent en utilisant Column.alias :

>>> 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()
-----------------------------------------
|"KEY1"  |"KEY2"  |"VALUE1"  |"VALUE2"  |
-----------------------------------------
|a       |a       |1         |3         |
|b       |b       |2         |4         |
-----------------------------------------

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

>>> df_lhs.join(df_rhs, df_lhs.col("key") == df_rhs.col("key"), lsuffix="_left", rsuffix="_right").show()
--------------------------------------------------
|"KEY_LEFT"  |"VALUE1"  |"KEY_RIGHT"  |"VALUE2"  |
--------------------------------------------------
|a           |1         |a            |3         |
|b           |2         |b            |4         |
--------------------------------------------------

Notez que ces exemples utilisent la méthode DataFrame.col pour spécifier les colonnes à utiliser dans la jointure. Voir Spécification des colonnes et des expressions pour d’autres façons de procéder.

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 qui utilisent un seul DataFrame pour effectuer une auto-jointure échouent 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()
8

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 qui doivent être sélectionnées.

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()
---------------------
|"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, voir 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()
-------------
|"A"  |"B"  |
-------------
|1    |3    |
-------------

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

    >>> # 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()
-------
|"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()
-----------------------
|"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()
---------------------
|"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()
[Row(status='Table 10tablename successfully created.')]
>>> session.sql("""insert into "10tablename" (id123, "3rdID", "id with space") values ('a', 'b', 'c')""").collect()
[Row(number of rows inserted=1)]
>>> df = session.table('"10tablename"')
>>> df.show()
---------------------------------------
|"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 :

>>> # The following calls are equivalent:
>>> df.select(col("id123")).collect()
[Row(ID123='a')]
>>> df.select(col("ID123")).collect()
[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 :

>>> # The following calls are equivalent:
>>> df.select(col("3rdID")).collect()
[Row(3rdID='b')]
>>> df.select(col("\"3rdID\"")).collect()
[Row(3rdID='b')]

>>> # The following calls are equivalent:
>>> df.select(col("id with space")).collect()
[Row(id with space='c')]
>>> df.select(col("\"id with space\"")).collect()
[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()
[Row(status='Table QUOTED successfully created.')]
>>> session.sql('''insert into quoted ("name_with_""air""_quotes", """column_name_quoted""") values ('a', 'b')''').collect()
[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()
[Row(name_with_"air"_quotes='a')]
>>> df_table.select("\"\"\"column_name_quoted\"\"\"").collect()
[Row("column_name_quoted"='b')]

N’oubliez pas que 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()
[Row(id with space='c')]

>>> 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()
-------------------------------
|"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 à nouveau 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.

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

Par exemple, dans le code ci-dessous, la méthode select renvoie un DataFrame qui ne contient que deux colonnes : name et serial_number. L’appel de la méthode filter sur ce DataFrame échoue car il utilise la colonne id, qui ne se trouve pas dans le DataFrame transformé.

>>> # This fails with the error "invalid identifier 'ID'."
>>> df_product_info = session.table("sample_product_data").select(col("name"), col("serial_number")).filter(col("id") == 1)
>>> try:
...   df_product_info.show()
... except SnowparkSQLException as e:
...   print(e.message)
000904 (42000): SQL compilation error: error line 1 at position 121
invalid identifier 'ID'

En revanche, le code suivant s’exécute avec succès car la méthode filter() est appelée sur un DataFrame qui contient toutes les colonnes de la table sample_product_data (y compris la colonne id) :

>>> # This succeeds because the DataFrame returned by the table() method
>>> # includes the "id" column.
>>> df_product_info = session.table("sample_product_data").filter(col("id") == 1).select(col("name"), col("serial_number"))
>>> df_product_info.show()
-------------------------------
|"NAME"     |"SERIAL_NUMBER"  |
-------------------------------
|Product 1  |prod-1           |
-------------------------------

N’oubliez pas que vous devrez peut-être 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.

>>> # 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 out:
>>> # 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. Notez que 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. Voir 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()

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()
-------------------------------------------------------------------------------------
|"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    |
-------------------------------------------------------------------------------------


>>> # Limit the number of rows to 20, rather than 10.
>>> df_products.show(20)
-------------------------------------------------------------------------------------
|"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    |
-------------------------------------------------------------------------------------

Remarque : 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.

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()

Sauvegarde des données dans une table

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

  1. Appelez la propriété write pour obtenir un objet DataFrameWriter.

  2. Appelez la méthode mode dans l’objet DataFrameWriter et précisez si vous voulez insérer des lignes ou mettre à jour des lignes dans la table. Cette méthode renvoie un nouvel objet DataFrameWriter qui est configuré avec le mode spécifié.

  3. 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.')]

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 :

  1. Appelez la méthode read de la classe Session pour accéder à un objet DataFrameReader.

  2. 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.

  3. 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.

  4. 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.

  5. 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.

  6. 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.

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()
[Row(status='Stage area MY_STAGE successfully created.')]
>>> stage_files_df = session.sql("ls @my_stage").collect()
>>> # Resume the operation of a warehouse.
>>> # Note that you must call the collect method in order to execute
>>> # the SQL statement.
>>> session.sql("alter warehouse if exists my_warehouse resume if suspended").collect()
[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()
[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()

>>> # 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'
Langage: Français