Faut-il une grappe de machines pour cela ?
Un même traitement de données peut se faire avec Spark, avec du SQL ou avec une simple boucle Python. Savoir choisir évite de lancer une grappe de machines pour rien.
Une équipe veut le chiffre d’affaires par pays. Une personne pense à Spark, une autre dit qu’une boucle de dix lignes suffirait. Les deux peuvent avoir raison : tout dépend de la taille des données. Vous faites ici le même travail de trois façons, puis vous décidez avec une raison.
À la fin de cette leçon
Remarque
Je peux faire un même traitement avec un DataFrame Spark, avec du SQL et avec du Python simple, et dire quand chacun convient.
Mots-clés
| Mot | Ce que ça veut dire | Exemple |
|---|---|---|
| DataFrame | une table de lignes et de colonnes nommées sur laquelle Spark travaille | |
| Spark SQL | poser une question à un DataFrame en SQL | |
| Redistribution | des lignes qui passent d’un travailleur à l’autre pour que les clés égales se retrouvent | Exchange |
| Plan physique | la liste des étapes que Spark va vraiment lancer | |
| En mémoire | assez petit pour tenir dans la mémoire d’une machine |
Les cellules que les étudiants remplissent restent vides dans cet aperçu.
Un travail, trois façons
Le travail est le chiffre d’affaires par pays : quantité fois prix, additionné pour chaque pays. Le labo a 24 commandes chargées sous le nom orders.
D’abord en expression de DataFrame Spark, puis en SQL Spark, puis en boucle Python simple sur le même fichier. Si les trois ne s’accordent pas, l’une se trompe.
La façon DataFrame
curl -s -d 'orders.groupBy("country").agg(F.round(F.sum(orders.quantity * orders.unit_price), 2).alias("revenue")).orderBy("country")' http://spark:8000/dfAffichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
La façon SQL
curl -s -d 'SELECT country, round(sum(quantity * unit_price), 2) AS revenue FROM orders GROUP BY country ORDER BY country' http://spark:8000/sqlAffichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
La façon Python simple
curl -s -d 'import csv, collections; c = collections.Counter(); [c.update({r["country"]: float(r["quantity"]) * float(r["unit_price"])}) for r in csv.DictReader(open("/tmp/lab/orders.csv"))]; sorted(c.items())' http://spark:8000/pyAffichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
Le même travail en boucle
# Le même traitement en boucle Python simple : le chiffre d'affaires par pays.
commandes = [["UK", 2, 25.0], ["Japan", 1, 150.0], ["UK", 1, 150.0], ["Japan", 4, 12.5]]
chiffre = {}
for commande in commandes:
pays = commande[0]
argent = commande[1] * commande[2]
if pays in chiffre:
chiffre[pays] = chiffre[pays] + argent
else:
chiffre[pays] = argent
print(chiffre)
# Ajoutez ["France", 3, 12.5] aux commandes et relancez.
Un DataFrame n’est pas une liste
Un DataFrame ressemble à une table que l’on parcourt, mais c’est un plan. Il ne tient pas les lignes dans votre programme, donc beaucoup d’opérations faites pour les listes échouent.
Demandez-lui sa taille avec count(), qui est une action. Demandez la longueur avec len() : Python refuse.
Demander sa longueur comme à une liste
curl -s -d 'len(orders)' http://spark:8000/dfAffichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
Demander sa taille comme à Spark
curl -s -d 'orders.count()' http://spark:8000/dfAffichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
Le plan montre où passent les données
Spark peut afficher le plan physique qu’il va lancer. Le mot à chercher est Exchange. C’est une redistribution : des lignes voyagent entre travailleurs pour que les clés égales se retrouvent ensemble.
Une redistribution est la partie coûteuse d’un gros traitement. Lire un plan dit où elle se trouve.
Un plan physique, tel que Spark l’affiche
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[country#19], functions=[count(1)])
+- Exchange hashpartitioning(country#19, 4), ENSURE_REQUIREMENTS, [plan_id=244]
+- HashAggregate(keys=[country#19], functions=[partial_count(1)])
+- InMemoryTableScan [country#19]
Lister les principaux opérateurs du plan
curl -s -d 'orders.groupBy("country").count()._jdf.queryExecution().simpleString()' http://spark:8000/df | grep -oE "Exchange|HashAggregate|FileScan"Affichée ici, non exécutée. Créez un compte gratuit pour appuyer sur Exécuter et la saisir dans le terminal de votre propre labo.
Une erreur fréquente
Attention
Beaucoup pensent que Spark est toujours le meilleur outil parce qu’il est le plus gros. Pour 24 lignes, il n’ajoute qu’un démarrage et un plan. Spark vaut le coup quand les données ne tiennent plus sur une machine, ou quand le travail doit être partagé.
Exemple travaillé : choisir un outil
| Les données | Un choix juste | Pourquoi |
|---|---|---|
| 24 lignes dans un fichier | Une boucle Python simple | Rien à répartir |
| 2 millions de lignes qui tiennent en mémoire | pandas, ou du SQL dans une base | Une machine le fait en quelques secondes |
| Des milliards de lignes sur plusieurs machines | Spark | Le travail doit être partagé |
| Un traitement lancé chaque nuit | Celui qui est écrit clairement et contrôlé | La même réponse chaque fois |
Les cellules que les étudiants remplissent restent vides dans cet aperçu.
Pour aller plus loin : une autre question
Calculez les unités vendues pour chaque produit avec la façon DataFrame, la façon SQL et la façon Python simple. S’accordent-elles ?
Indice : groupez par product et additionnez quantity. Puis affichez le plan de votre ligne DataFrame et trouvez son Exchange.