Utilisation de DataFrames dans Snowpark Python¶

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

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

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

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

  Copy

• 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
  

  Copy

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

  Copy

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

  Copy

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

  Copy

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

  Copy

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

  Copy

• 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
  

  Copy

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

  Copy

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

  Copy

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

  Copy

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.

# 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         |
-------------------------------

# 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         |
-------------------------------

# 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         |
-------------------------------

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

# 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         |
-----------------------------------------------------

# 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         |
-----------------------------------------

# 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         |
--------------------------------------------------

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.

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

# 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   |
---------------------

# 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    |
---------------------

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

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

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

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

[Row(ID123='a')]

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

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

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

[Row(3rdID='b')]

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

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

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

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

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"'

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

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

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

# 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 :

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

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

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

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 :

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

      Copy

   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)
      

      Copy

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

   Copy

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

   Copy

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

   Copy

   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.

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

from snowflake.snowpark.functions import col

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

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

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

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

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

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

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

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

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

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

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

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

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

----------------------------------------------------------------------------------------------------------------------------------------------------------
|"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"                       |       |       |        |         |                                   |                                     |
|    }                                      |       |       |        |         |                                   |                                     |
|  ]                                        |       |       |        |         |                                   |                                     |
|}                                          |       |       |        |         |                                   |                                     |
----------------------------------------------------------------------------------------------------------------------------------------------------------

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

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¶

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.

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

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

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 :

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.