Introduction
Certains systèmes livrent des enregistrements sous forme de lignes CSV ; d’autres fournissent des objets JSON avec des champs nommés. Le commerçant fournit maintenant des fichiers d’événements contenant un libellé de type et des valeurs liées aux commandes. Vos questions de traitement sont connues : que signifient les champs, quels enregistrements correspondent à des commandes et quelles colonnes le résultat doit-il exposer ?
JSON Lines stocke un objet JSON complet par ligne. Contrairement au CSV, les noms de champs accompagnent chaque objet, et un objet peut en contenir un autre. Vous lirez d’abord des enregistrements plats avec des types explicites, sélectionnerez ensuite les champs d’un objet client imbriqué simple et appliquerez enfin ce schéma à une seconde livraison.
Ce sont des fichiers finis lus avec spark.read ; il s’agit donc toujours de traitement par lots. Un nom de fichier contenant events ne transforme pas une application en traitement de flux. L’environnement et les données d’exercice sont indépendants ; vous avez besoin des notions précédentes de schéma explicite, select, filter et alias, ainsi que des bases de Python. Aucun Kafka, Hadoop, Hive, configuration de streaming, tableau ou récursion complexe n’est nécessaire. Utilisez le Terminal.
Lire du JSON Lines plat et sélectionner les événements de commande
Dans cette étape, vous transposerez la lecture avec schéma explicite du CSV au JSON. Examinez le fichier fourni :
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Chaque ligne est un objet entouré d’accolades. Les noms entre guillemets doubles identifient les champs ; un deux-points sépare un nom de sa valeur. Les nombres comme quantity sont des valeurs JSON numériques, tandis que order_date est une chaîne de date entre guillemets que le DateType déclaré interprétera. Il n’y a ni ligne d’en-tête CSV ni tableau englobant.
L’order_id, la quantité, le prix et la date du premier objet décrivent le même type de valeurs qu’une commande CSV. Le fichier contient aussi un événement d’inventaire ; lire un enregistrement valide ne signifie pas qu’il appartient à un résultat de commandes. Le schéma explicite ci-dessous déclare les champs plats. json(path) lit ce format, tandis que le filter connu sélectionne event_type == order.
Enregistrez le script complet avec un here-document entre apostrophes et exécutez-le avec l’interpréteur PySpark préparé :
cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt
cat /home/labex/project/output/flat-json-orders.txt
L’affichage initial doit inclure les trois événements. Sous Order events, vous devez obtenir uniquement j1/2/10.00/2026-10-14 et j2/1/25.00/2026-10-14. L’enregistrement d’inventaire a été lu correctement, mais exclu par la condition de type. Le schéma doit montrer une quantité integer, un prix decimal et order_date de type date.
Le lecteur utilise les champs JSON nommés avec les types déclarés ; contrairement à un enregistrement CSV simple, l’ordre des champs dans chaque objet JSON ne définit pas l’identité de la colonne. Le schéma définit toujours le contrat prévu.
Sélectionner les champs d’un objet client imbriqué
Dans cette étape, vous lirez un niveau de structure supplémentaire. Examinez le fichier d’événements clients :
cat /home/labex/project/data/customer-events.jsonl
La valeur customer est un autre objet, contenant customer_id et region. Dans le schéma Spark, cela devient une structure (struct), un champ contenant des champs enfants nommés. Ce n’est ni un tableau ni une seconde table à joindre.
Déclarez cette structure avec un StructType interne. col("customer.region") sélectionne le champ enfant par son chemin pointé ; alias("region") donne au résultat sélectionné un nom de colonne simple. Appliquez la même idée à l’identifiant client :
cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt
cat /home/labex/project/output/nested-json-orders.txt
Le schéma complet doit montrer customer comme une struct avec deux enfants string. La table sélectionnée doit contenir j3/c1/East/2/10.00/2026-10-15 et j4/c2/West/1/25.00/2026-10-15. Son schéma final expose customer_id et region comme des colonnes string ordinaires à côté de la quantité, du prix et de la date.
Vous avez sélectionné des valeurs déjà présentes dans chaque événement de commande. Aucune table de référence ni jointure n’était requise. La représentation imbriquée originale reste disponible dans events, tandis que selected fournit une vue plate plus simple pour les opérations suivantes.
Réutiliser le schéma imbriqué sur un autre fichier
Dans cette étape, vous changerez la livraison tout en conservant la même structure. Lisez d’abord le nouveau fichier :
cat /home/labex/project/data/new-customer-events.jsonl
Prévoyez la sortie : une commande j5 pour c2 dans West, de quantité 3 et de prix 12.50 le 16 octobre. Le lecteur et la sélection imbriquée peuvent rester identiques ; seul le chemin d’entrée change. Exécutez cette version complète :
cat > /home/labex/project/read_json_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = SparkSession.builder.appName("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt
cat /home/labex/project/output/new-json-orders.txt
Vous devez retrouver le résultat prévu d’une ligne et les mêmes types sélectionnés. Un nouveau fichier ne nécessite pas d’inventer un nouveau schéma lorsque la structure convenue est inchangée.
Les trois lectures ont utilisé spark.read sur des fichiers existants et se sont terminées avec leurs scripts. Elles ne se sont pas abonnées à de futurs messages. Vous avez appris à varier un format d’entrée sans changer simultanément le modèle d’exécution. Vous reviendrez ensuite à une livraison de commandes volontairement imparfaite pour définir des règles explicites de nettoyage.
Résumé
Vous avez lu du JSON Lines statique avec des types scalaires explicites, sélectionné les événements de commande et aplati une structure client simple avec des chemins de colonnes nommés. Une seconde livraison a correctement utilisé le même schéma. Vous traiterez ensuite les enregistrements de commandes manquants, dupliqués et invalides selon des règles de nettoyage claires.
