Démarrer avec PySpark#

Spark est une plateforme open-source de traitement de données volumineuses. Au fil des années, Spark s'est imposé comme l'outil de référence pour l'ingénierie de données. Dans ce guide ultime, je vous présente PySpark, l'API Python de Spark.

import warnings
warnings.filterwarnings("ignore")  # Ignore warnings coming from Arrow optimizations.

Importer PySpark#

import pyspark

pyspark.__version__

'3.5.1'

Créer une session Spark#

Nous devons importer la classe SparkSession depuis le module pyspark.sql

from pyspark.sql import SparkSession

# Créer une session Spark
spark = SparkSession.builder \
    .appName("Get Started") \
    .getOrCreate()

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
24/05/12 17:51:07 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
24/05/12 17:51:07 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.

Créer un Dataframe#

Nous allons préparer une petit dataset, la liste des colonnes et les enregistrements comme liste de tuples

columns = ["id", "product_name", "product_category", "amount", "quantity"]
data = [
    (123, "Product1", "Category1", 45.67, 8),
    (124, "Product2", "Category2", 78.23, 3),
    (125, "Product3", "Category3", 32.45, 6),
    (126, "Product4", "Category1", 91.12, 5),
    (127, "Product5", "Category2", 64.78, 7)
]

Nous allons créer le dataframe à partir du dataset

df = spark.createDataFrame(data, schema=columns)

Afficher le DataFrame#

On fait appel à la méthode .show() pour afficher le dataframe.

df.show()

+---+------------+----------------+------+--------+
| id|product_name|product_category|amount|quantity|
+---+------------+----------------+------+--------+
|123|    Product1|       Category1| 45.67|       8|
|124|    Product2|       Category2| 78.23|       3|
|125|    Product3|       Category3| 32.45|       6|
|126|    Product4|       Category1| 91.12|       5|
|127|    Product5|       Category2| 64.78|       7|
+---+------------+----------------+------+--------+

Sélection#

df.select("id").show()

+---+
| id|
+---+
|123|
|124|
|125|
|126|
|127|
+---+

Tri#

df.sort("quantity").show()

+---+------------+----------------+------+--------+
| id|product_name|product_category|amount|quantity|
+---+------------+----------------+------+--------+
|124|    Product2|       Category2| 78.23|       3|
|126|    Product4|       Category1| 91.12|       5|
|125|    Product3|       Category3| 32.45|       6|
|127|    Product5|       Category2| 64.78|       7|
|123|    Product1|       Category1| 45.67|       8|
+---+------------+----------------+------+--------+

Filtrage#

df.filter("product_category = 'Category1'").show()

+---+------------+----------------+------+--------+
| id|product_name|product_category|amount|quantity|
+---+------------+----------------+------+--------+
|123|    Product1|       Category1| 45.67|       8|
|126|    Product4|       Category1| 91.12|       5|
+---+------------+----------------+------+--------+

Il existe cette synthaxe

df.filter(df["product_category"] == "Category1").show()

+---+------------+----------------+------+--------+
| id|product_name|product_category|amount|quantity|
+---+------------+----------------+------+--------+
|123|    Product1|       Category1| 45.67|       8|
|126|    Product4|       Category1| 91.12|       5|
+---+------------+----------------+------+--------+

Agrégation#

agg_df = df.groupby("product_category").agg({"id": "count", "amount": "sum", "quantity": "sum"})

agg_df.show()

+----------------+------------------+-------------+---------+
|product_category|       sum(amount)|sum(quantity)|count(id)|
+----------------+------------------+-------------+---------+
|       Category1|136.79000000000002|           13|        2|
|       Category2|            143.01|           10|        2|
|       Category3|             32.45|            6|        1|
+----------------+------------------+-------------+---------+

Manipulation avec SQL#

Spark offre la possibilité de manipuler les données avec SQL, c'est l'une de ses forces. Les professionnels des données qui utilisent SQL au quotidien s'y retrouvent assez facilement.

df.createOrReplaceTempView("lu_products")

sql_query = "SELECT * FROM lu_products WHERE id &gt; 125"

result_df = spark.sql(sql_query)
result_df.show()

+---+------------+----------------+------+--------+
| id|product_name|product_category|amount|quantity|
+---+------------+----------------+------+--------+
|126|    Product4|       Category1| 91.12|       5|
|127|    Product5|       Category2| 64.78|       7|
+---+------------+----------------+------+--------+

Fermer la session Spark#

Une fois notre travail terminé, nous devons fermer la session.

spark.stop()

Dans la prochaine section, nous verrons comment importer des données depuis des fichiers.