Utilisation de DataFrames dans Snowpark Scala¶

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 Colonne en appelant la fonction col dans l’objet com.snowflake.snowpark.functions.

// Import the col function from the functions object.
import com.snowflake.snowpark.functions._

val dfProductInfo = session.table("sample_product_data").select(col("id"), col("name"))
dfProductInfo.show()

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 an SQL SELECT statement.
  df.filter(col("id") === 20)
  

  Copy

  // Specify the equivalent of "WHERE a + b < 10"
  // in an SQL SELECT statement.
  df.filter((col("a") + col("b")) < 10)
  

  Copy

• 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 an SQL SELECT statement.
  df.select((col("b") * 10) as "c")
  

  Copy

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

  // Specify the equivalent of "X JOIN Y on X.a_in_X = Y.b_in_Y"
  // in an SQL SELECT statement.
  dfX.join(dfY, col("a_in_X") === col("b_in_Y"))
  

  Copy

// Create a DataFrame that joins two other DataFrames (dfLhs and dfRhs).
// Use the DataFrame.col method to refer to the columns used in the join.
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("key") === dfRhs.col("key")).select(dfLhs.col("value").as("L"), dfRhs.col("value").as("R"))

// Create a DataFrame that joins two other DataFrames (dfLhs and dfRhs).
// Use the DataFrame.apply method to refer to the columns used in the join.
// Note that dfLhs("key") is shorthand for dfLhs.apply("key").
val dfJoined = dfLhs.join(dfRhs, dfLhs("key") === dfRhs("key")).select(dfLhs("value").as("L"), dfRhs("value").as("R"))

Utilisation de Shorthand pour un objet de colonne¶

Au lieu d’utiliser la fonction col, vous pouvez faire référence à une colonne de l’une des manières suivantes :

• Utilisez un signe dollar devant le nom de la colonne entre guillemets simples ($"column_name").

• Utilisez une apostrophe (un guillemet simple) devant le nom de la colonne non mise entre guillemets simples ('column_name).

Pour ce faire, importez les noms de l’objet implicits après avoir créé un objet Session :

val session = Session.builder.configFile("/path/to/properties").create

// Import this after you create the session.
import session.implicits._

// Use the $ (dollar sign) shorthand.
val df = session.table("T").filter($"id" === 10).filter(($"a" + $"b") < 10)

// Use ' (apostrophe) shorthand.
val df = session.table("T").filter('id === 10).filter(('a + 'b) < 10).select('b * 10)

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. 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"))
df.select(col("ID123"))

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

val 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’identificateur :

// The following calls are equivalent:
df.select(col("3rdID"))
df.select(col("\"3rdID\""))

// The following calls are equivalent:
df.select(col("id with space"))
df.select(col("\"id with space\""))

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 :

describe table quoted;
+------------------------+ ...
| name                   | ...
|------------------------+ ...
| name_with_"air"_quotes | ...
| "column_name_quoted"   | ...
+------------------------+ ...

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

val dfTable = session.table("quoted")
dfTable.select("\"name_with_\"\"air\"\"_quotes\"").show()
dfTable.select("\"\"\"column_name_quoted\"\"\"").show()

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"))
df.select(col("ID WITH SPACE"))

Utilisation de littéraux comme objets de colonne¶

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

// Import for the lit and col functions.
import com.snowflake.snowpark.functions._

// 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.filter(col("num_items").gt(lit(5))).show()

Si le littéral est une valeur à virgule flottante ou double en Scala (par exemple, 0.05 est traité comme un double par défaut), la bibliothèque Snowpark génère du SQL qui convertit implicitement la valeur en type de données Snowpark correspondant (par exemple, 0.05::DOUBLE). Cela peut produire une valeur approximative qui diffère du nombre exact spécifié.

Par exemple, le code suivant n’affiche aucune ligne correspondante, alors que le filtre (qui correspond aux valeurs supérieures ou égales à 0.05) devrait correspondre aux lignes de DataFrame :

// Create a DataFrame that contains the value 0.05.
val df = session.sql("select 0.05 :: Numeric(5, 2) as a")

// Applying this filter results in no matching rows in the DataFrame.
df.filter(col("a") <= lit(0.06) - lit(0.01)).show()

Le problème est que lit(0.06) et lit(0.01) produisent des valeurs approximatives pour 0.06 et 0.01, et non les valeurs exactes.

Pour éviter ce problème, vous pouvez utiliser l’une des approches suivantes :

• Option 1 : convertir le littéral dans le type Snowpark que vous voulez utiliser. Par exemple, pour utiliser un NUMBER avec une précision de 5 et une échelle de 2 :

  df.filter(col("a") <= lit(0.06).cast(new DecimalType(5, 2)) - lit(0.01).cast(new DecimalType(5, 2))).show()
  

  Copy

• Option 2 : convertir la valeur dans le type que vous souhaitez utiliser avant de la transmettre à la fonction lit. Par exemple, si vous voulez utiliser le type BigDecimal :

  df.filter(col("a") <= lit(BigDecimal(0.06)) - lit(BigDecimal(0.01))).show()
  

  Copy

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 Column.cast et transmettez un objet de type à partir du paquet com.snowflake.snowpark.types package. 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.
import com.snowflake.snowpark.functions._
// Import for the DecimalType class..
import com.snowflake.snowpark.types._

val decimalValue = lit(0.05).cast(new 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.

val dfProductInfo = session.table("sample_product_data").filter(col("id") === 1).select(col("name"), col("serial_number"))
dfProductInfo.show()

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'."
val dfProductInfo = session.table("sample_product_data").select(col("name"), col("serial_number")).filter(col("id") === 1)

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.
val dfProductInfo = session.table("sample_product_data").filter(col("id") === 1).select(col("name"), col("serial_number"))
dfProductInfo.show()

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.

Limitation du nombre de lignes dans un DataFrame¶

Pour limiter le nombre de lignes dans un DataFrame, vous pouvez utiliser la méthode de transformation DataFrame.limit.

L’API de Snowpark fournit également des méthodes d’action pour récupérer et imprimer un nombre limité de lignes :

• la méthode d’action DataFrame.first (pour exécuter la requête et renvoyer les n premières lignes)

• la méthode d’action DataFrame.show (pour exécuter la requête et imprimer les n premières lignes)

Ces méthodes ajoutent effectivement une clause LIMIT à l’instruction SQL qui est exécutée.

Comme expliqué dans les notes sur l’utilisation de LIMIT, les résultats sont non-déterministes à moins que vous ne spécifiiez un ordre de tri (ORDER BY) en conjonction avec LIMIT.

Pour conserver la clause ORDER BY avec la clause LIMIT (par exemple, pour que ORDER BY ne soit pas dans une sous-requête séparée), vous devez appeler la méthode qui limite les résultats sur le DataFrame retourné par la méthode sort.

Par exemple, si vous enchaînez des appels de méthode :

// Limit the number of rows to 5, sorted by parent_id.
var dfSubset = df.sort(col("parent_id")).limit(5);

// Return the first 5 rows, sorted by parent_id.
var arrayOfRows = df.sort(col("parent_id")).first(5)

// Print the first 5 rows, sorted by parent_id.
df.sort(col("parent_id")).show(5)

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 méthode schema. Cette méthode renvoie un objet StructType qui contient un Array 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.
val tableSchema = session.table("sample_product_data").schema
println("Schema for sample_product_data: " + tableSchema);

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 méthode schema, puis la méthode names sur l’objet StructType renvoyé pour obtenir un ArraySeq de noms de colonnes. Les noms sont normalisés dans les StructType retournés par la méthode schema.

// Create a DataFrame containing the "id" and "3rd" columns.
val dfSelectedColumns = session.table("sample_product_data").select(col("id"), col("3rd"))
// Print out the names of the columns in the schema. This prints out:
//   ArraySeq(ID, "3rd")
println(dfSelectedColumns.schema.names.toSeq)

Joindre des DataFrames¶

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

Les sections suivantes expliquent comment utiliser des DataFrames pour effectuer une jointure :

• Configuration des données d’exemple pour les jointures

• Spécifier les colonnes pour la jointure

• Exécution d’une jointure naturelle

• Spécification du type de jointure

• Joindre plusieurs tables

• Exécution d’une auto-jointure

create or replace table sample_a (
  id_a integer,
  name_a varchar,
  value integer
);
insert into sample_a (id_a, name_a, value) values
  (10, 'A1', 5),
  (40, 'A2', 10),
  (80, 'A3', 15),
  (90, 'A4', 20)
;
create or replace table sample_b (
  id_b integer,
  name_b varchar,
  id_a integer,
  value integer
);
insert into sample_b (id_b, name_b, id_a, value) values
  (4000, 'B1', 40, 10),
  (4001, 'B2', 10, 5),
  (9000, 'B3', 80, 15),
  (9099, 'B4', null, 200)
;
create or replace table sample_c (
  id_c integer,
  name_c varchar,
  id_a integer,
  id_b integer
);
insert into sample_c (id_c, name_c, id_a, id_b) values
  (1012, 'C1', 10, null),
  (1040, 'C2', 40, 4000),
  (1041, 'C3', 40, 4001)
;

// Create a DataFrame that joins the DataFrames for the tables
// "sample_a" and "sample_b" on the column named "id_a".
val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a") === dfRhs.col("id_a"))
dfJoined.show()

----------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |
----------------------------------------------------------------------
|10      |A1        |5        |4001    |B2        |10      |5        |
|40      |A2        |10       |4000    |B1        |40      |10       |
|80      |A3        |15       |9000    |B3        |80      |15       |
----------------------------------------------------------------------

val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("id_a") === dfRhs.col("id_a"))
val dfSelected = dfJoined.select(dfLhs.col("value").as("LeftValue"), dfRhs.col("value").as("RightValue"))
dfSelected.show()

------------------------------
|"LEFTVALUE"  |"RIGHTVALUE"  |
------------------------------
|5            |5             |
|10           |10            |
|15           |15            |
------------------------------

--------------------------------------------------------------------------------------------------
|"l_ZSz7_ID_A"  |"NAME_A"  |"l_ZSz7_VALUE"  |"ID_B"  |"NAME_B"  |"r_heec_ID_A"  |"r_heec_VALUE"  |
--------------------------------------------------------------------------------------------------
|10             |A1        |5               |4001    |B2        |10             |5               |
|40             |A2        |10              |4000    |B1        |40             |10              |
|80             |A3        |15              |9000    |B3        |80             |15              |
--------------------------------------------------------------------------------------------------

val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfJoined = dfLhs.naturalJoin(dfRhs)
dfJoined.show()

---------------------------------------------------
|"ID_A"  |"VALUE"  |"NAME_A"  |"ID_B"  |"NAME_B"  |
---------------------------------------------------
|10      |5        |A1        |4001    |B2        |
|40      |10       |A2        |4000    |B1        |
|80      |15       |A3        |9000    |B3        |
---------------------------------------------------

// Create a DataFrame that performs a left outer join on
// "sample_a" and "sample_b" on the column named "id_a".
val dfLhs = session.table("sample_a")
val dfRhs = session.table("sample_b")
val dfLeftOuterJoin = dfLhs.join(dfRhs, dfLhs.col("id_a") === dfRhs.col("id_a"), "left")
dfLeftOuterJoin.show()

----------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |
----------------------------------------------------------------------
|40      |A2        |10       |4000    |B1        |40      |10       |
|10      |A1        |5        |4001    |B2        |10      |5        |
|80      |A3        |15       |9000    |B3        |80      |15       |
|90      |A4        |20       |NULL    |NULL      |NULL    |NULL     |
----------------------------------------------------------------------

val dfFirst = session.table("sample_a")
val dfSecond  = session.table("sample_b")
val dfThird = session.table("sample_c")
val dfJoinThreeTables = dfFirst.join(dfSecond, dfFirst.col("id_a") === dfSecond.col("id_a")).join(dfThird, dfFirst.col("id_a") === dfThird.col("id_a"))
dfJoinThreeTables.show()

------------------------------------------------------------------------------------------------------------
|"ID_A"  |"NAME_A"  |"VALUE"  |"ID_B"  |"NAME_B"  |"ID_A"  |"VALUE"  |"ID_C"  |"NAME_C"  |"ID_A"  |"ID_B"  |
------------------------------------------------------------------------------------------------------------
|10      |A1        |5        |4001    |B2        |10      |5        |1012    |C1        |10      |NULL    |
|40      |A2        |10       |4000    |B1        |40      |10       |1040    |C2        |40      |4000    |
|40      |A2        |10       |4000    |B1        |40      |10       |1041    |C3        |40      |4001    |
------------------------------------------------------------------------------------------------------------

// This fails because columns named "id" and "parent_id"
// are in the left and right DataFrames in the join.
val df = session.table("sample_product_data");
val dfJoined = df.join(df, col("id") === col("parent_id"))

// This fails because columns named "id" and "parent_id"
// are in the left and right DataFrames in the join.
val df = session.table("sample_product_data");
val dfJoined = df.join(df, df("id") === df("parent_id"))

Exception in thread "main" com.snowflake.snowpark.SnowparkClientException:
  Joining a DataFrame to itself can lead to incorrect results due to ambiguity of column references.
  Instead, join this DataFrame to a clone() of itself.

// Create a DataFrame object for the "sample_product_data" table for the left-hand side of the join.
val dfLhs = session.table("sample_product_data")
// Clone the DataFrame object to use as the right-hand side of the join.
val dfRhs = dfLhs.clone()

// Create a DataFrame that joins the two DataFrames
// for the "sample_product_data" table on the
// "id" and "parent_id" columns.
val dfJoined = dfLhs.join(dfRhs, dfLhs.col("id") === dfRhs.col("parent_id"))
dfJoined.show()

// Create a DataFrame that performs a self-join on a DataFrame
// using the column named "key".
val df = session.table("sample_product_data");
val dfJoined = df.join(df, Seq("key"))

Exécution d’une action de manière synchrone¶

Pour exécuter une action de manière synchrone, appelez l’une des méthodes d’action suivantes :

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.
val dfProducts = session.table("sample_product_data")

// Send the query to the server for execution and
// print the count of rows in the table.
println("Rows returned: " + dfProducts.count())

Vous pouvez également appeler des méthodes d’action pour :

• Exécuter une requête sur une table et renvoyer les résultats.

• Exécuter une requête et imprimer les résultats dans la console.

Remarque : si vous appelez la méthode schema pour obtenir les définitions des colonnes dans DataFrame, vous n’avez pas besoin d’appeler une méthode d’action.

Exécution d’une action de manière asynchrone¶

Pour effectuer une action de manière asynchrone, appelez la méthode async pour retourner un objet « acteur asynchrone » (par exemple DataFrameAsyncActor), et appelez une méthode d’action asynchrone dans cet objet.

Ces méthodes d’action d’un objet acteur asynchrone renvoient un objet TypedAsyncJob que vous pouvez utiliser pour vérifier l’état de l’action asynchrone et récupérer les résultats de l’action.

Les sections suivantes expliquent comment effectuer des actions de manière asynchrone et vérifier les résultats.

• Comprendre le déroulement de base des actions asynchrones

• Spécification du nombre maximum de secondes à attendre

• Accès à une requête asynchrone par ID

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

// Execute the query asynchronously.
// This call does not block.
val asyncJob = df.async.collect()
// Check if the query has completed execution.
println(s"Is query ${asyncJob.getQueryId()} done? ${asyncJob.isDone()}")
// Get an Array of Rows containing the results, and print the results.
// Note that getResult is a blocking call.
val results = asyncJob.getResult()
results.foreach(println)

// Create a DataFrame for the "sample_product_data" table.
val dfProducts = session.table("sample_product_data")

// Execute the query asynchronously.
// This call does not block.
val asyncJob = df.async.count()
// Check if the query has completed execution.
println(s"Is query ${asyncJob.getQueryId()} done? ${asyncJob.isDone()}")
// Print the count of rows in the table.
// Note that getResult is a blocking call.
println("Rows returned: " + asyncJob.getResult())

// Wait a maximum of 10 seconds for the query to complete before retrieving the results.
val results = asyncJob.getResult(10)

val asyncJob = session.createAsyncJob(myQueryId)
// Check if the query has completed execution.
println(s"Is query ${asyncJob.getQueryId()} done? ${asyncJob.isDone()}")
// If you need to retrieve the results, call getRows to return an Array of Rows containing the results.
// Note that getRows is a blocking call.
val rows = asyncJob.getRows()
rows.foreach(println)

Renvoi de toutes les lignes¶

Pour renvoyer toutes les lignes en une seule fois, appelez la méthode DataFrame.collect. Cette méthode renvoie un tableau d’objets Row. Pour récupérer les valeurs de la ligne, appelez la méthode getType (par exemple getString, getInt, etc.).

Par exemple :

import com.snowflake.snowpark.functions_

val rows = session.table("sample_product_data").select(col("name"), col("category_id")).sort(col("name")).collect()
for (row <- rows) {
  println(s"Name: ${row.getString(0)}; Category ID: ${row.getInt(1)}")
}

Renvoie d’un itérateur pour les lignes¶

Si vous voulez utiliser un itérateur pour effectuer une itération sur les objets Row dans les résultats, appelez DataFrame.toLocalIterator. Si la quantité de données dans les résultats est importante, la méthode charge les lignes par morceaux pour éviter de charger toutes les lignes en mémoire en même temps.

Par exemple :

import com.snowflake.snowpark.functions_

while (rowIterator.hasNext) {
  val row = rowIterator.next()
  println(s"Name: ${row.getString(0)}; Category ID: ${row.getInt(1)}")
}

Retourner les n premières lignes¶

Pour renvoyer les n premières lignes, appelez la méthode DataFrame.first, en indiquant le nombre de lignes à renvoyer.

Comme expliqué dans Limitation du nombre de lignes dans un DataFrame, les résultats sont non déterministes. Si vous voulez que les résultats soient déterministes, appelez cette méthode sur un DataFrame trié (df.sort().first()).

Par exemple :

import com.snowflake.snowpark.functions_

val df = session.table("sample_product_data")
val rows = df.sort(col("name")).first(5)
rows.foreach(println)

Mise à jour des lignes dans une table¶

Pour la méthode update , on transmet un Map qui associe les colonnes à mettre à jour et les valeurs correspondantes à affecter à ces colonnes. update renvoie un objet UpdateResult qui contient le nombre de lignes mises à jour. (Voir UpdateResult.)

Par exemple, pour remplacer les valeurs de la colonne nommée count par la valeur 1 :

val updatableDf = session.table("sample_product_data")
val updateResult = updatableDf.update(Map("count" -> lit(1)))
println(s"Number of rows updated: ${updateResult.rowsUpdated}")

L’exemple ci-dessus utilise le nom de la colonne pour identifier la colonne. Vous pouvez également utiliser une expression de colonne :

val updateResult = updatableDf.update(Map(col("count") -> lit(1)))

Si la mise à jour ne doit être effectuée que lorsqu’une condition est remplie, vous pouvez spécifier cette condition comme argument. Par exemple, pour remplacer les valeurs de la colonne nommée count pour les lignes dans lesquelles la colonne category_id a la valeur 20 :

val updateResult = updatableDf.update(Map(col("count") -> lit(1)), col("category_id") === 20)

Si vous devez baser la condition sur une jointure avec un objet DataFrame différent, vous pouvez transmettre ce DataFrame en tant qu’argument et utiliser ce DataFrame dans la condition. Par exemple, pour remplacer les valeurs de la colonne nommée count pour les lignes dans lesquelles la colonne category_id correspond à category_id dans les DataFrame dfParts :

val updatableDf = session.table("sample_product_data")
val dfParts = session.table("parts")
val updateResult = updatableDf.update(Map(col("count") -> lit(1)), updatableDf("category_id") === dfParts("category_id"), dfParts)

Suppression de lignes dans une table¶

Pour la méthode delete vous pouvez spécifier une condition qui identifie les lignes à supprimer, et vous pouvez baser cette condition sur une jointure avec un autre DataFrame. delete renvoie un objet DeleteResult qui contient le nombre de lignes supprimées. (Voir DeleteResult.)

Par exemple, pour supprimer les lignes qui ont la valeur 1 dans la colonne category_id :

val updatableDf = session.table("sample_product_data")
val deleteResult = updatableDf.delete(updatableDf("category_id") === 1)
println(s"Number of rows deleted: ${deleteResult.rowsDeleted}")

Si la condition fait référence à des colonnes dans un DataFrame différent, transmettez ce DataFrame comme deuxième argument. Par exemple, pour supprimer les lignes dans lesquelles la colonne category_id correspond à category_id dans DataFrame dfParts, transmettez dfParts comme second argument :

val updatableDf = session.table("sample_product_data")
val deleteResult = updatableDf.delete(updatableDf("category_id") === dfParts("category_id"), dfParts)
println(s"Number of rows deleted: ${deleteResult.rowsDeleted}")

Fusion de lignes dans une table¶

Pour insérer, mettre à jour et supprimer des lignes dans une table en fonction des valeurs d’une deuxième table ou d’une sous-requête (l’équivalent de la commande MERGE en SQL), procédez comme suit :

1. Dans l’objet Updatable de la table dans laquelle vous voulez fusionner les données, appelez la méthode merge en transmettant l’objet DataFrame de l’autre table et l’expression de la colonne pour la condition de jointure.

   Le résultat est un objet MergeBuilder que vous pouvez utiliser pour spécifier les actions à entreprendre (par exemple, insérer, mettre à jour ou supprimer) sur les lignes qui correspondent et celles qui ne correspondent pas. Voir (MergeBuilder.)

2. En utilisant l’objet MergeBuilder :

   • Pour spécifier la mise à jour ou la suppression qui doit être effectuée sur les lignes correspondantes, appelez la méthode whenMatched.

     Si vous devez spécifier une condition supplémentaire pour que les lignes soient mises à jour ou supprimées, vous pouvez transmettre une expression de colonne pour cette condition.

     Cette méthode renvoie un objet MatchedClauseBuilder que vous pouvez utiliser pour spécifier l’action à exécuter. (Voir MatchedClauseBuilder.)

     Appelez la méthode update ou delete dans l’objet MatchedClauseBuilder pour spécifier l’action de mise à jour ou de suppression qui doit être exécutée sur les lignes correspondantes. Ces méthodes renvoient un objet MergeBuilder que vous pouvez utiliser pour spécifier des clauses supplémentaires.

   • Pour spécifier l’insertion qui doit être effectuée lorsque des lignes ne correspondent pas, appelez la méthode whenNotMatched.

     Si vous devez spécifier une condition supplémentaire pour l’insertion de lignes, vous pouvez transmettre une expression de colonne pour cette condition.

     Cette méthode renvoie un objet NotMatchedClauseBuilder que vous pouvez utiliser pour spécifier l’action à exécuter. (Voir NotMatchedClauseBuilder.)

     Appelez la méthode insert dans l’objet NotMatchedClauseBuilder pour spécifier l’action d’insertion qui doit être exécutée lorsque les lignes ne correspondent pas. Ces méthodes renvoient un objet MergeBuilder que vous pouvez utiliser pour spécifier des clauses supplémentaires.

3. Lorsque vous avez fini de spécifier les insertions, mises à jour et suppressions à effectuer, appelez la méthode collect de l’objet MergeBuilder pour effectuer les insertions, mises à jour et suppressions spécifiées sur la table.

   collect renvoie un objet MergeResult qui contient le nombre de lignes qui ont été insérées, mises à jour et supprimées. (Voir MergeResult.)

L’exemple suivant insère une ligne avec les colonnes id et value de la table source dans la table target si la table target ne contient pas de ligne avec un ID correspondant :

val mergeResult = target.merge(source, target("id") === source("id"))
                      .whenNotMatched.insert(Seq(source("id"), source("value")))
                      .collect()

L’exemple suivant met à jour une ligne de la table target avec la valeur de la colonne value de la ligne de la table source qui a le même ID :

val mergeResult = target.merge(source, target("id") === source("id"))
                      .whenMatched.update(Map("value" -> source("value")))
                      .collect()

Chargement et téléchargement de fichiers dans une zone de préparation¶

Pour charger et télécharger des fichiers dans une zone de préparation, utilisez l’objet FileOperation :

• Chargement de fichiers vers une zone de préparation

• Téléchargement de fichiers à partir d’une zone de préparation

Utilisation des flux d’entrée pour charger et télécharger des données dans une zone de préparation¶

Pour utiliser les flux d’entrée afin de charger des données dans un fichier sur une zone de préparation et de télécharger des données d’un fichier sur une zone de préparation, utilisez les méthodes uploadStream et downloadStream de l’objet FileOperation :

• Utilisation d’un flux d’entrée pour télécharger des données vers un fichier dans une zone de préparation

• Utilisation d’un flux d’entrée pour télécharger des données à partir d’un fichier sur une zone de préparation

import java.io.InputStream
...
val compressData = true
val pathToFileOnStage = "@myStage/path/file"
session.file.uploadStream(pathToFileOnStage, new ByteArrayInputStream(fileContent.getBytes()), compressData)

import java.io.InputStream
...
val isDataCompressed = true
val pathToFileOnStage = "@myStage/path/file"
val is = session.file.downloadStream(pathToFileOnStage, isDataCompressed)

Configuration d’un DataFrame pour des fichiers dans une zone de préparation¶

Cette section explique comment configurer un DataFrame pour des fichiers dans une zone de préparation Snowflake. Une fois que vous avez créé ce DataFrame, vous pouvez utiliser le DataFrame pour :

• récupérer des données des fichiers

• copier les données des fichiers dans une table

Pour configurer un DataFrame pour des fichiers dans une zone de préparation Snowflake, utilisez la classe DataFrameReader :

1. Vérifiez que vous disposez des privilèges suivants :

   • Privilèges pour accéder aux fichiers dans la zone de préparation.

   • Un des éléments suivants :

     • Privilèges CREATE TABLE sur le schéma, si vous prévoyez de spécifier des options de copie qui déterminent la façon dont les données sont copiées à partir des fichiers en zone de préparation.

     • Sinon, privilèges CREATE FILE FORMAT sur le schéma.

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

3. Si les fichiers sont au format CSV, décrivez les champs du fichier. Pour ce faire :

   1. Créez un objet StructType composé d’une séquence 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 paquet com.snowflake.snowpark.types).

      • Indique si le champ peut être « Null » ou non.

      Par exemple :

      import com.snowflake.snowpark.types._
      
      val schemaForDataFile = StructType(
          Seq(
              StructField("id", StringType, true),
              StructField("name", StringType, true)))
      

      Copy

   3. Appelez la méthode schema dans l’objet DataFrameReader, en transmettant l’objet StructType.

      Par exemple :

      var dfReader = session.read.schema(schemaForDataFile)
      

      Copy

      La méthode 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.

4. 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 la méthode DataFrameReader.option ou la méthode DataFrameReader.options.

   Saisissez le nom et la valeur de l’option que vous souhaitez définir. Vous pouvez définir les types d’options suivants :

   • Les options de format de fichier décrites dans 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.

   dfReader = dfReader.option("field_delimiter", ";").option("COMPRESSION", "NONE")
   

   Copy

   La méthode option renvoie un objet DataFrameReader qui est configuré avec l’option spécifiée.

   Pour définir plusieurs options, vous pouvez soit enchaîner les appels à la méthode option (comme dans l’exemple ci-dessus), soit appeler la méthode DataFrameReader.options, en transmettant un Map des noms et valeurs des options.

5. Appelez la méthode correspondant au format des fichiers. Vous pouvez appeler l’une des méthodes suivantes :

   • DataFrameReader.avro

   • DataFrameReader.csv

   • DataFrameReader.json

   • DataFrameReader.orc

   • DataFrameReader.parquet

   • DataFrameReader.xml

   Lorsque vous appelez ces méthodes, transmettez l’emplacement de zone de préparation des fichiers à lire. Par exemple :

   val df = dfReader.csv("@s3_ts_stage/emails/data_0_0_0.csv")
   

   Copy

   Pour spécifier plusieurs fichiers qui commencent par le même préfixe, spécifiez le préfixe après le nom de la zone de préparation. Par exemple, pour charger les fichiers qui ont le préfixe csv_ à partir de la zone de préparation @mystage :

   val df = dfReader.csv("@mystage/csv_")
   

   Copy

   Les méthodes correspondant au format d’un fichier renvoient un objet CopyableDataFrame pour ce fichier. CopyableDataFrame étend DataFrame et fournit des méthodes supplémentaires pour exploiter les données dans des fichiers en zone de préparation.

6. Appelez une méthode d’action pour :

   • récupérer les données des fichiers, ou

   • copier les données des fichiers dans une table

   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.

Chargement de données à partir de fichiers dans un DataFrame¶

Après avoir configuré un DataFrame pour des fichiers dans une zone de préparation, vous pouvez charger les données des fichiers dans DataFrame :

1. 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 nommé data.json dans la zone de préparation nommée mystage :

   val df = session.read.json("@mystage/data.json").select(col("$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.

2. Appelez la méthode DataFrame.collect pour charger les données. Par exemple :

   val results = df.collect()
   

   Copy

Copier les données d’un fichier dans une table¶

Après avoir configuré un DataFrame pour des fichiers dans une zone de préparation, vous pouvez appeler la méthode CopyableDataFrame.copyInto pour copier les données dans une table. Cette méthode exécute la commande COPY INTO <table>.

Par exemple, le code suivant charge les données du fichier CSV spécifié par myFileStage dans la table mytable. Comme les données se trouvent dans un fichier CSV, le code doit également décrire les champs du fichier. Pour ce faire, l’exemple appelle la méthode DataFrameReader.schema et transmet un objet StructType (csvFileSchema) contenant une séquence d’objets StructField qui décrivent les champs.

val df = session.read.schema(csvFileSchema).csv(myFileStage)
df.copyInto("mytable")

Sauvegarde d’un DataFrame dans des fichiers sur une zone de préparation¶

Si vous devez enregistrer un DataFrame dans des fichiers dans une zone de préparation, vous pouvez appeler la méthode DataFrameWriter correspondant au format du fichier (par exemple, la méthode csv pour écrire dans un fichier CSV), en indiquant l’emplacement de la zone de préparation où les fichiers doivent être enregistrés. Ces méthodes DataFrameWriter exécutent la commande COPY INTO <emplacement>.

Pour enregistrer le contenu d’un DataFrame dans des fichiers dans une zone de préparation :

1. Appelez la méthode DataFrame.write pour obtenir un objet DataFrameWriter. Par exemple, pour obtenir l’objet DataFrameWriter pour un DataFrame qui représente la table nommée sample_product_data :

   dfWriter = session.table("sample_product_data").write
   

   Copy

2. Si vous voulez écraser le contenu du fichier (si le fichier existe), appelez la méthode DataFrameWriter.mode, en transmettant SaveMode.Overwrite.

   Sinon, par défaut, DataFrameWriter signale une erreur si le fichier spécifié dans la zone de préparation existe déjà.

   La méthode mode renvoie le même objet DataFrameWriter configuré avec le mode spécifié.

   Par exemple, pour spécifier que DataFrameWriter doit écraser le fichier dans la zone de préparation :

   dfWriter = dfWriter.mode(SaveMode.Overwrite)
   

   Copy

3. Si vous devez spécifier des informations supplémentaires sur la manière dont les données doivent être enregistrées (par exemple, que les données doivent être compressées ou que vous souhaitez utiliser un point-virgule pour délimiter les champs dans un fichier CSV), appelez la méthode DataFrameWriter.option ou la méthode DataFrameWriter.options.

   Saisissez le nom et la valeur de l’option que vous souhaitez définir. Vous pouvez définir les types d’options suivants :

   • Les options de format de fichier décrites dans la documentation sur COPY INTO <emplacement>.

   • Les options de copie décrites dans la documentation sur COPY INTO <emplacement>.

   • PARTITIONBY ou HEADER.

   Notez que vous ne pouvez pas utiliser la méthode option pour définir les options suivantes :

   • L’option de type de format TYPE.

   • Option de copie OVERWRITE. Pour définir cette option, appelez plutôt la méthode mode (comme indiqué à l’étape précédente).

   L’exemple suivant configure l’objet DataFrameWriter pour enregistrer des données dans un fichier CSV sous forme non compressée, en utilisant un point-virgule (plutôt qu’une virgule) comme délimiteur de champ.

   dfWriter = dfWriter.option("field_delimiter", ";").option("COMPRESSION", "NONE")
   

   Copy

   La méthode option renvoie un objet DataFrameWriter qui est configuré avec l’option spécifiée.

   Pour définir plusieurs options, vous pouvez enchaîner les appels à la méthode option (comme dans l’exemple ci-dessus), ou appeler la méthode DataFrameWriter.options, en transmettant un Map des noms et valeurs des options.

4. Pour obtenir des détails sur chaque fichier sauvegardé, définissez DETAILED_OUTPUT l’option de copie sur TRUE.

   Par défaut, DETAILED_OUTPUT est FALSE, ce qui signifie que la méthode renvoie une seule ligne de sortie contenant les champs "rows_unloaded", "input_bytes" et "output_bytes".

   Lorsque vous définissez DETAILED_OUTPUT sur TRUE, la méthode renvoie une ligne de sortie pour chaque fichier enregistré. Chaque ligne contient les champs FILE_NAME, FILE_SIZE, et ROW_COUNT.

5. Appelez la méthode correspondant au format du fichier pour enregistrer les données dans le fichier. Vous pouvez appeler l’une des méthodes suivantes :

   • DataFrameWriter.csv

   • DataFrameWriter.json

   • DataFrameWriter.parquet

   Lorsque vous appelez ces méthodes, transmettez l’emplacement de la zone de préparation du fichier où les données doivent être écrites (par exemple, @mystage).

   Par défaut, la méthode enregistre les données dans des noms de fichiers avec le préfixe data_ (par exemple, @mystage/data_0_0_0.csv). Si vous voulez que les fichiers soient nommés avec un préfixe différent, spécifiez le préfixe après le nom de la zone de préparation. Par exemple :

   val writeFileResult = dfWriter.csv("@mystage/saved_data")
   

   Copy

   Cet exemple enregistre le contenu du DataFrame dans des fichiers qui commencent par le préfixe saved_data (par exemple, @mystage/saved_data_0_0_0.csv).

6. Vérifiez l’objet WriteFileResult retourné pour obtenir des informations sur la quantité de données écrites dans le fichier.

   À partir de l’objet WriteFileResult, vous pouvez accéder à la sortie produite par la commande COPY INTO <emplacement> :

   • Pour accéder aux lignes de sortie sous la forme d’un tableau d’objets Row, utilisez le membre valeur rows.

   • Pour déterminer les champs présents dans les lignes, utilisez le membre valeur schema, qui est un StructType décrivant les champs de la ligne.

   Par exemple, pour imprimer les noms des champs et des valeurs dans les lignes de sortie :

   val writeFileResult = dfWriter.csv("@mystage/saved_data")
   for ((row, index) <- writeFileResult.rows.zipWithIndex) {
     (writeFileResult.schema.fields, writeFileResult.rows(index).toSeq).zipped.foreach {
       (structField, element) => println(s"${structField.name}: $element")
     }
   }
   

   Copy

L’exemple suivant utilise un DataFrame pour enregistrer le contenu de la table nommée car_sales dans des fichiers JSON avec le préfixe saved_data dans la zone de préparation @mystage (par exemple @mystage/saved_data_0_0_0.json). L’exemple de code :

• Écrase le fichier, si le fichier existe déjà dans la zone de préparation.

• Renvoie une sortie détaillée sur l’opération de sauvegarde.

• Enregistre les données non compressées.

Enfin, l’exemple de code imprime chaque champ et chaque valeur dans les lignes de sortie renvoyées :

val df = session.table("car_sales")
val writeFileResult = df.write.mode(SaveMode.Overwrite).option("DETAILED_OUTPUT", "TRUE").option("compression", "none").json("@mystage/saved_data")
for ((row, index) <- writeFileResult.rows.zipWithIndex) {
  println(s"Row: $index")
  (writeFileResult.schema.fields, writeFileResult.rows(index).toSeq).zipped.foreach {
    (structField, element) => println(s"${structField.name}: $element")
  }
}

Parcours de données semi-structurées¶

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

• Utilisez Column.apply(« <field_name> ») pour retourner un objet Column pour un champ dans un OBJECT (ou un VARIANT qui contient un OBJECT).

• Utilisez Column.apply(<index>) pour renvoyer un objet Column pour un élément dans un ARRAY (ou un VARIANT qui contient un ARRAY).

Comme mentionné dans Utilisation de la méthode apply pour faire référence à une colonne, vous pouvez omettre le nom de la méthode apply :

col("column_name")("field_name")
col("column_name")(index)

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

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

Le code imprime la sortie suivante :

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

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

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

val df = session.table("car_sales")
df.select(col("src")("salesperson")("name")).show()

Le code imprime la sortie suivante :

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

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

val df = session.table("car_sales")
df.select(col("src")("vehicle")(0)).show()
df.select(col("src")("vehicle")(0)("price")).show()

Le code imprime la sortie suivante :

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

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

Comme alternative à la méthode apply, vous pouvez utiliser les fonctions get, get_ignore_case ou get_path si le nom du champ ou les éléments du chemin sont irréguliers et rendent difficile l’utilisation des méthodes Column.apply.

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

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

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

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

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

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

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

// Import the objects for the data types, including StringType.
import com.snowflake.snowpark.types._
...
val df = session.table("car_sales")
df.select(col("src")("salesperson")("id")).show()
df.select(col("src")("salesperson")("id").cast(StringType)).show()

Le code imprime la sortie suivante :

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

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

Aplatissement d’un tableau d’objets en lignes¶

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

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

Si vous transmettez ce chemin à la fonction flatten :

val df = session.table("car_sales")
df.flatten(col("src")("customer")).show()

cette méthode renvoie un DataFrame :

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

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

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

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

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

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