Introduction
Un commerçant reçoit des commandes et souhaite les transformer en informations utiles. Avant de calculer des rapports de ventes, vous avez besoin d’un moyen de représenter les lignes, de décrire leurs colonnes et de demander à l’outil de traitement d’exécuter du travail. Vous commencerez avec trois enregistrements assez petits pour être vérifiés à l’œil.
Le big data décrit des charges de données dont le volume, le rythme d’arrivée ou les besoins de traitement peuvent dépasser une approche pratique sur une seule machine. Apache Spark est un moteur de traitement de données capable de répartir le travail entre les machines d’un cluster. PySpark est son interface Python. Un DataFrame représente des lignes avec des colonnes nommées et typées et prend en charge des opérations comme le filtrage et l’agrégation. Spark traite les données ; un fichier ou système de stockage séparé fournit les entrées et sorties persistantes. Ce cours commence par de petits exemples pour vous permettre de raisonner sur l’exactitude avant l’échelle.
Vous créerez une SparkSession, afficherez un DataFrame, examinerez son schéma et son nombre de lignes, puis ajouterez un enregistrement avant de relancer le script. L’environnement préparé utilise local[2] : Spark s’exécute sur cette seule machine avec deux fils d’exécution. Cela enseigne de véritables API Spark, tandis que le déploiement distribué est un sujet distinct. Vous avez besoin des bases Python sur les variables, listes et appels de fonctions ; le parcours Python peut vous aider pour ces prérequis. Aucun pandas, Hadoop, Hive, Java, installation ou fichier d’un lab précédent n’est nécessaire. Utilisez le Terminal.
Exécuter un petit script DataFrame complet
Dans cette étape, vous exécuterez un script complet transformant trois enregistrements Python en table Spark. Commencez dans votre espace de travail :
cd /home/labex/project
Une SparkSession est le point d’entrée pour créer des DataFrames et accéder aux API de données structurées de Spark. Le générateur définit un nom d’application, sélectionne l’exécution locale et crée la session. local[2] autorise deux fils d’exécution sur cette machine ; il ne crée pas deux machines de cluster. setLogLevel réduit les journaux d’information suivants pour faciliter le repérage des sorties pédagogiques.
Dans le code Python, from pyspark.sql import SparkSession rend cette classe installée disponible pour votre programme. Un point accède à un membre nommé, et les parenthèses appellent une méthode : df.show() demande à ce DataFrame d’afficher ses lignes. Le générateur enchaîne plusieurs appels ; les parenthèses qui l’entourent permettent à l’expression d’occuper plusieurs lignes. Vous pouvez lire chaque ligne comme un paramètre avant que getOrCreate() obtienne la session.
La liste orders contient trois tuples, des groupes de valeurs entre parenthèses. Chaque tuple fournit une ligne dans le même ordre que les colonnes déclarées. La chaîne de schéma nomme chaque colonne et son type : des identifiants et produits textuels, suivis d’une quantité entière. createDataFrame utilise ces enregistrements et ce schéma. show demande à Spark d’afficher les lignes, et stop libère la session une fois le travail terminé.
Enregistrez le script complet. Le > du shell écrit le fichier nommé ; le marqueur PYTHON entre apostrophes commence un here-document qui préserve littéralement le texte Python inclus. Copiez jusqu’à la ligne PYTHON finale :
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
spark.stop()
PYTHON
La commande bigdata-python préparée sélectionne l’environnement Python où PySpark est installé. Exécutez le fichier comme un programme Python normal en enregistrant sa sortie ordinaire localement :
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt
cat /home/labex/project/output/first-run.txt
Vous devez obtenir une table avec les colonnes order_id, product et quantity, contenant s1/notebook/2, s2/pen/1 et s3/folder/3. Des avertissements de démarrage ou messages de progression peuvent encore apparaître dans le terminal ; ils sont séparés de la table enregistrée. Aucun ordre de lignes n’est garanti sans tri explicite ; identifiez donc les enregistrements par leurs identifiants de commande.
Votre programme Python est le pilote (driver) : il crée la session et demande à Spark d’effectuer du travail. Dans un cluster, des processus exécuteurs exécutent des tâches sur des portions de données. Ici, l’exécution se produit localement sur la même machine. Vous disposez d’un programme fonctionnel sans avoir à configurer de cluster.
Examiner le schéma et compter les lignes
Dans cette étape, vous distinguerez ce qu’une colonne peut contenir du nombre d’enregistrements de la table. Un schéma liste les noms de colonnes, les types de données et la possibilité de valeurs nulles. Une colonne entière représente des nombres entiers adaptés aux calculs de quantités ; le texte possède d’autres opérations. Nullable signifie que le champ autorise une valeur manquante, et non que chaque ligne en possède une.
Ajoutez printSchema pour examiner cette description et count pour calculer le nombre de lignes. Voici le fichier mis à jour complet pour qu’aucun état interactif ne soit caché :
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt
cat /home/labex/project/output/schema-and-count.txt
La table doit toujours contenir trois enregistrements. En dessous, repérez :
root
|-- order_id: string (nullable = true)
|-- product: string (nullable = true)
|-- quantity: integer (nullable = true)
Row count: 3
printSchema décrit le DataFrame. show et count sont des actions qui demandent à Spark de produire un résultat visible ou une valeur. Spark peut différer une grande partie du traitement jusqu’à ce qu’une action demande une réponse ; les labs suivants expliqueront cette exécution paresseuse avec les transformations et les plans. Un nombre de 3 signifie trois enregistrements de commandes, et non la somme des quantités, qui répondrait à une autre question métier.
Chaque invocation démarre un nouveau processus Python et crée sa propre SparkSession. Le fichier crée explicitement df ; il ne dépend pas d’un shell interactif ayant créé spark ou df auparavant.
Prévoir et tester un enregistrement supplémentaire
Dans cette étape, vous changerez les données tout en conservant leur structure. Prévoyez l’effet de l’ajout de s4/notebook/1 : le nombre de commandes devient quatre, le schéma reste le même et les trois enregistrements originaux restent présents. La nouvelle ligne est une commande, même si notebook apparaît déjà dans une autre commande.
Remplacez le script par cette version, qui ajoute uniquement cet enregistrement :
cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = (
SparkSession.builder
.appName("FirstOrderTable")
.master("local[2]")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
orders = [
("s1", "notebook", 2),
("s2", "pen", 1),
("s3", "folder", 3),
("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)
df.show()
df.printSchema()
print("Row count:", df.count())
spark.stop()
PYTHON
Exécutez à nouveau le fichier dans un nouveau processus :
bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt
cat /home/labex/project/output/expanded-run.txt
Trouvez les quatre identifiants de commandes et Row count: 4. Le schéma doit toujours décrire les trois mêmes noms et types de colonnes. Répéter un nom de produit ne regroupe pas automatiquement les commandes et ne supprime pas les doublons ; ces opérations doivent être demandées explicitement dans les leçons suivantes.
Les fichiers précédents first-run.txt et schema-and-count.txt restent comme preuves de votre version à trois lignes. Le fichier exécutable décrit maintenant la version à quatre lignes ; une autre exécution indépendante peut donc reproduire ce résultat. Vous avez effectué un petit cycle modifier-prévoir-exécuter-vérifier, qui vous aidera à juger des traitements de données plus importants plutôt que de simplement croire qu’un programme terminé avec succès est correct.
Résumé
Vous avez exécuté un véritable script PySpark avec sa propre SparkSession, créé un DataFrame typé et vérifié ses enregistrements affichés, son schéma et son nombre de lignes. Ajouter un enregistrement a changé le nombre de lignes tout en préservant le schéma. Vous passerez ensuite des enregistrements écrits dans Python à un fichier CSV avec un schéma explicite.
