Architecture et concepts fondamentaux

Dans l'écosystème Apache Spark, les DataFrames et Spark SQL constituant des abstractions de haut niveau optimisées. Un DataFrame est une collection de données distribuée organisée en colonnes nommées. Bien qu'il soit conceptuellement équivalent à une table de base de données relationnelle, il bénéficie en arrière-plan des optimisations du moteur Catalyst. La différence majeure avec les RDD (Resilient Distributed Datasets) réside dans la présence d'un schéma explicite définissant le nom et le type de chaque colonne.

Initialisation de l'environnement

Depuis Spark 2.0, SparkSession est le point d'entrée unifié pour interagir avec les DataFrames et Spark SQL, remplaçant les anciens contextes tels que SparkContext et SQLContext.

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DataProcessingApp") \
    .master("local[*]") \
    .config("spark.sql.warehouse.dir", "/user/hive/warehouse") \
    .enableHiveSupport() \
    .getOrCreate()

Méthodes d'instanciation des DataFrames

1. À partir de collections Python (Listes ou Tuples)

Il est possible de convertir directemant des structures de données locales en DataFrame distribué en fournissant le schéma sous forme de liste.

dataset = [("Alice", 34, "Engineering"), ("Bob", 28, "Marketing"), ("Charlie", 45, "Sales")]
schema_cols = ["employee_name", "age", "department"]

df_employees = spark.createDataFrame(dataset, schema=schema_cols)
df_employees.show()

2. Conversion depuis un RDD

Les RDD existants peuvent être transformés en DataFrames pour bénéficier des optimisations SQL et d'un traitement plus rapide.

rdd_source = spark.sparkContext.parallelize([("Paris", 2100000), ("Lyon", 510000)])
df_cities = spark.createDataFrame(rdd_source, schema=["city_name", "population"])
df_cities.printSchema()

3. Interopérabilité avec Pandas

Les DataFrames Pandas en mémoire peuvent être convertis en DataFrames Spark pour un traitement distribué. Les noms de colonnes natifs de Pandas sont automatiquement conservés.

import pandas as pd

pdf = pd.DataFrame({"product_id": [101, 102], "category": ["Electronics", "Home"]})
df_spark_from_pandas = spark.createDataFrame(pdf)
df_spark_from_pandas.show()

4. Ingestion de fichiers plats

L'objet DataFrameReader permet de charger divers formats de fichiers depuis des systèmes distribués (HDFS, S3) ou locaux.

Format CSV : Le lecteur permet de déduire le schéma ou de l'appliquer manuellement. L'option header gère la présence de noms de colonnes dans le fichier, évitant les conflits de schéma lors de l'ingestion.

df_csv = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("hdfs://namenode:8020/data/transactions.csv")

Format JSON : Les fichiers JSON (avec un objet par ligne) sont mappés directement sur la structure hiérarchique du DataFrame.

df_json = spark.read.json("s3a://bucket-name/logs/events.json")

Format Parquet : Format colonnaire natif de Spark, conservant nativement les métadonnées de typage et de compression.

df_parquet = spark.read.parquet("/data/warehouse/user_metrics/")

Fichiers texte bruts : La méthode text() charge chaque ligne du fichier dans une colonne unique de type String nommée par défaut value.

df_text = spark.read.text("/var/log/application/app.log")

5. Connexion aux bases de données relationnelles

Tables Hive : Si le support Hive est activé lors de l'initialisation, les tables peuvent être requêtées directement.

df_hive = spark.table("analytics.customer_churn")

Sources JDBC (ex: MySQL) : La connexion nécessite l'URL du serveur, le nom de la table cible et un dictionnaire de propriétés incluant les identifiants et le pilote JDBC.

jdbc_url = "jdbc:mysql://db-host:3306/production"
connection_props = {
    "user": "data_engineer",
    "password": "secure_password",
    "driver": "com.mysql.cj.jdbc.Driver"
}

df_mysql = spark.read.jdbc(url=jdbc_url, table="orders", properties=connection_props)

Interaction entre l'API DataFrame et Spark SQL

Spark SQL est un module construit au-dessus de l'API DataFrame. Fondamentalement, un DataFrame agit comme une table relationnelle. Pour exécuter des requêtes SQL standard, la structure distribuée doit d'abord être enregistrée en tant que vue temporaire.

df_employees.createOrReplaceTempView("vw_employees")

sql_query = """
    SELECT department, AVG(age) as average_age 
    FROM vw_employees 
    GROUP BY department
"""
df_sql_result = spark.sql(sql_query)
df_sql_result.show()

Bien que Spark SQL offre une syntaxe familière et flexible pour les analystes de données, l'API DataFrame garantit une vérification des types à la compilation et une meilleure intégration avec les environnements de développement. Sous le capot, les deux approches génèrent le même plan d'exécution logique et physique optimisé par le moteur Catalyst.