Aller au contenu

Import et Export#

Spark peut lire des données depuis des sources externes tels que les fichiers CSV, JSON, Parquet ou encore HDFS.

Pré-requis#

Afin d'accéder à la base de données Postgres, nous aurons besoin du driver Postgres. Rendez-vous sur la page de téléchargement Download pgJDBC.

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

Importer PySpark#

import os
import urllib
import pyspark
from pyspark.sql import SparkSession
from pyspark.context import SparkContext

Créer une session Spark#

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

# 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:34:15 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
24/05/12 17:34:15 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.

Importer depuis un fichier CSV#

Récupérez le dataset StudentsPerformance.csv ici.

URL = "https://raw.githubusercontent.com/joekakone/datasets/master/datasets/StudentsPerformance.csv"
FILENAME = "StudentsPerformance.csv"

urllib.request.urlretrieve(url=URL, filename=FILENAME)

('StudentsPerformance.csv', <http.client.HTTPMessage at 0x7fac0fe66aa0>)

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

df = spark.read.csv(path=FILENAME, header=True, inferSchema=True)
df.show(10)

+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
|gender|race/ethnicity|parental level of education|       lunch|test preparation course|math score|reading score|writing score|
+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
|female|       group B|          bachelor's degree|    standard|                   none|        72|           72|           74|
|female|       group C|               some college|    standard|              completed|        69|           90|           88|
|female|       group B|            master's degree|    standard|                   none|        90|           95|           93|
|  male|       group A|         associate's degree|free/reduced|                   none|        47|           57|           44|
|  male|       group C|               some college|    standard|                   none|        76|           78|           75|
|female|       group B|         associate's degree|    standard|                   none|        71|           83|           78|
|female|       group B|               some college|    standard|              completed|        88|           95|           92|
|  male|       group B|               some college|free/reduced|                   none|        40|           43|           39|
|  male|       group D|                high school|free/reduced|              completed|        64|           64|           67|
|female|       group B|                high school|free/reduced|                   none|        38|           60|           50|
+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
only showing top 10 rows

Nous allons créer le dataframe à partir du dataset

Afficher le schéma#

Le schéma d'un DataFrame en PySpark est une représentation tabulaire qui expose la structure des données, incluant les noms des colonnes, les types de données associés à chaque colonne, et d'autres informations pertinentes. La méthode printSchema permet d'afficher le schéma.

df.printSchema()

root
 |-- gender: string (nullable = true)
 |-- race/ethnicity: string (nullable = true)
 |-- parental level of education: string (nullable = true)
 |-- lunch: string (nullable = true)
 |-- test preparation course: string (nullable = true)
 |-- math score: integer (nullable = true)
 |-- reading score: integer (nullable = true)
 |-- writing score: integer (nullable = true)

Lire un fichier à partir d'une URL#

from pyspark import SparkFiles

sc = spark.sparkContext

sc.addFile(URL)

path = SparkFiles.get(os.path.split(URL)[-1])

df = spark.read.csv("file://" + path, header=True, inferSchema=True, sep=",")

df.show()

+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
|gender|race/ethnicity|parental level of education|       lunch|test preparation course|math score|reading score|writing score|
+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
|female|       group B|          bachelor's degree|    standard|                   none|        72|           72|           74|
|female|       group C|               some college|    standard|              completed|        69|           90|           88|
|female|       group B|            master's degree|    standard|                   none|        90|           95|           93|
|  male|       group A|         associate's degree|free/reduced|                   none|        47|           57|           44|
|  male|       group C|               some college|    standard|                   none|        76|           78|           75|
|female|       group B|         associate's degree|    standard|                   none|        71|           83|           78|
|female|       group B|               some college|    standard|              completed|        88|           95|           92|
|  male|       group B|               some college|free/reduced|                   none|        40|           43|           39|
|  male|       group D|                high school|free/reduced|              completed|        64|           64|           67|
|female|       group B|                high school|free/reduced|                   none|        38|           60|           50|
|  male|       group C|         associate's degree|    standard|                   none|        58|           54|           52|
|  male|       group D|         associate's degree|    standard|                   none|        40|           52|           43|
|female|       group B|                high school|    standard|                   none|        65|           81|           73|
|  male|       group A|               some college|    standard|              completed|        78|           72|           70|
|female|       group A|            master's degree|    standard|                   none|        50|           53|           58|
|female|       group C|           some high school|    standard|                   none|        69|           75|           78|
|  male|       group C|                high school|    standard|                   none|        88|           89|           86|
|female|       group B|           some high school|free/reduced|                   none|        18|           32|           28|
|  male|       group C|            master's degree|free/reduced|              completed|        46|           42|           46|
|female|       group C|         associate's degree|free/reduced|                   none|        54|           58|           61|
+------+--------------+---------------------------+------------+-----------------------+----------+-------------+-------------+
only showing top 20 rows

Exporter vers un fichier CSV#

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

OUTPUT_FILE = "StudentsPerformanceOutput.csv"

df.write.csv(path=OUTPUT_FILE, mode="overwrite")

Exporter vers un fichier Parquet#

Apache Parquet est un format de fichier de données open source en colonnes, conçu pour stocker et récupérer des données avec une grande efficacité. Pour en savoir plus, merci de consulter ce lien Parquet, qu'est-ce que c'est ?

OUTPUT_FILE = "StudentsPerformance.parquet"

df.write.parquet(path=OUTPUT_FILE, mode="overwrite", partitionBy="race/ethnicity")

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.