Utiliser Spark pour travailler avec des fichiers de données

Après avoir configuré un notebook et l’attacher à un cluster, vous pouvez utiliser Spark pour lire et traiter des fichiers de données. Spark prend en charge un large éventail de formats, tels que CSV, JSON, Parquet, ORC, Avro et Delta, et Databricks fournit des connecteurs intégrés pour accéder aux fichiers stockés dans l’espace de travail, dans Azure Data Lake ou Stockage Blob, ou dans d’autres systèmes externes.

Le flux de travail suit généralement trois étapes :

1. Lisez un fichier dans un DataFrame Spark à l’aide de spark.read avec le format et le chemin appropriés. Lors de la lecture de formats de texte brut comme CSV ou JSON, Spark peut déduire le schéma (noms de colonnes et types de données), mais cela est parfois lent ou non fiable. Une meilleure pratique en production consiste à définir le schéma explicitement afin que les données soient chargées de manière cohérente et efficace.

2. Explorez et transformez le DataFrame à l’aide d’opérations SQL ou DataFrame (par exemple, filtrage de lignes, sélection de colonnes, agrégation de valeurs).

3. Réécrire les résultats dans le stockage dans un format choisi.

L’utilisation de fichiers dans Spark est conçue pour être cohérente entre les jeux de données petits et volumineux. Le même code utilisé pour tester un petit fichier CSV fonctionne également sur des jeux de données beaucoup plus volumineux, car Spark distribue le travail sur le cluster. Cela facilite le scale-up de l’exploration rapide vers le traitement des données plus complexe.

Chargement des données dans un dataframe

Prenons un exemple hypothétique pour voir comment utiliser un dataframe pour travailler avec des données. Supposons que vous disposiez des données suivantes dans un fichier texte délimité par des virgules nommé products.csv stocké dans un volume de catalogue Unity :

ProductID,ProductName,Category,ListPrice
771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

Dans un notebook Spark, vous pouvez utiliser le code PySpark suivant pour charger les données dans un dataframe et afficher les 10 premières lignes :

%python
df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/products.csv',
    format='csv',
    header=True,
    inferSchema=True
)
display(df.limit(10))

La %python ligne au début est appelée magic et indique à Spark que la langue utilisée dans cette cellule est Python. Voici le code Scala équivalent pour l’exemple de données des produits :

%scala
val df = spark.read.format("csv").option("header", "true").option("inferSchema", "true").load("/Volumes/my_catalog/my_schema/my_volume/products.csv")
display(df.limit(10))

La commande magic %scala est utilisée pour spécifier Scala.

Les deux exemples présentés précédemment produisent une sortie comme suit :

Spécification d’un schéma de dataframe

Dans l’exemple précédent, la première ligne du fichier CSV contenait les noms de colonnes et Spark déduisait le type de données de chaque colonne des données qu’elle contient, car le code définissait les options inferSchema et header. Vous pouvez également spécifier un schéma explicite pour les données, ce qui est utile lorsque les noms de colonne ne sont pas inclus dans le fichier de données, comme cet exemple CSV :

771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

L’exemple PySpark suivant montre comment spécifier un schéma pour que le dataframe soit chargé à partir d’un fichier appelé product-data.csv dans ce format :

from pyspark.sql.types import *
from pyspark.sql.functions import *

productSchema = StructType([
    StructField("ProductID", IntegerType()),
    StructField("ProductName", StringType()),
    StructField("Category", StringType()),
    StructField("ListPrice", FloatType())
    ])

df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/product-data.csv',
    format='csv',
    schema=productSchema,
    header=False)
display(df.limit(10))

Les résultats seraient une fois de plus similaires à :

Filtrage et regroupement des dataframes

Vous pouvez utiliser les méthodes de la classe Dataframe pour filtrer, trier, regrouper et manipuler les données qu’elle contient. Par exemple, l’exemple de code suivant utilise la select méthode pour récupérer les colonnes ProductID et ListPrice à partir du dataframe df contenant des données de produit dans l’exemple précédent :

pricelist_df = df.select("ProductID", "ListPrice")

Les résultats de cet exemple de code devraient ressembler à ceci :

En commun avec la plupart des méthodes de manipulation de données, select retourne un nouvel objet de trame de données.

Vous pouvez « chaîner » les méthodes ensemble pour effectuer une série de manipulations qui entraînent un dataframe transformé. Par exemple, cet exemple de code enchaîne les méthodes select et where pour créer un nouveau dataframe contenant les colonnes ProductName et ListPrice pour les produits dans les catégories Mountain Bikes ou Road Bikes :

bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)

Les résultats de cet exemple de code devraient ressembler à ceci :

Pour regrouper et agréger des données, vous pouvez utiliser la méthode et les groupBy fonctions d’agrégation. Par exemple, le code PySpark suivant compte le nombre de produits de chaque catégorie :

counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)

Les résultats de cet exemple de code devraient ressembler à ceci :

Utilisation d’expressions SQL dans Spark

L’API Dataframe fait partie d’une bibliothèque Spark appelée Spark SQL, qui permet aux analystes Données d’utiliser des expressions SQL pour interroger et manipuler des données.

Création d’objets de base de données dans le catalogue Spark

Le catalogue Spark est un metastore pour les objets de données relationnelles tels que les vues et les tables. Le runtime Spark peut utiliser le catalogue pour intégrer du code écrit dans n’importe quel langage pris en charge par Spark avec des expressions SQL qui peuvent être plus naturelles pour certains analystes de données ou développeurs.

L’une des méthodes les plus simples pour rendre les données d’un dataframe disponibles pour pouvoir les interroger dans le catalogue Spark consiste à créer une vue temporaire, comme illustré dans l’exemple de code suivant :

df.createOrReplaceTempView("products")

Une vue est temporaire, ce qui signifie qu’elle est automatiquement supprimée à la fin de la session active. Vous pouvez également créer des tables persistantes dans le catalogue pour définir une base de données pouvant être interrogée à l’aide de Spark SQL.

Utilisation de l’API Spark SQL pour interroger des données

Vous pouvez utiliser l’API Spark SQL dans le code écrit dans n’importe quel langage pour interroger les données du catalogue. Par exemple, le code PySpark suivant utilise une requête SQL pour retourner les données de la vue produits en tant que dataframe.

bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
                      FROM products \
                      WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)

Les résultats de l’exemple de code ressembleraient au tableau suivant :

Utilisation du code SQL

L’exemple précédent a montré comment utiliser l’API Spark SQL pour incorporer des expressions SQL dans le code Spark. Dans un notebook, vous pouvez également utiliser la commande magic %sql pour exécuter le code SQL qui interroge les objets du catalogue, comme suit :

%sql

SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category

L’exemple de code SQL retourne un jeu de résultats qui s’affiche automatiquement dans le notebook sous forme de tableau, comme celui ci-dessous :