Un chiffre par pays, trois milliards de lignes
Chaque lundi, une enseigne veut un seul chiffre : ses ventes par pays. La table des commandes compte trois milliards de lignes, réparties sur vingt machines.
Vous ne voulez pas écrire vingt programmes. Vous voulez décrire le résultat une fois, et laisser le système répartir le travail.
À la fin de cette leçon
Remarque
Je peux exprimer un traitement de données sous la forme d'un pipeline Spark.
Mots clés
| Mot | Ce que ça veut dire | Exemple |
|---|---|---|
| DataFrame | une table de lignes et de colonnes nommées sur laquelle Spark travaille | |
| Transformation | une étape qui fabrique une nouvelle table à partir d'une autre | filter |
| Action | une étape qui exécute le plan et donne une réponse | count |
| Paresseux | qui attend qu'on demande une réponse avant de travailler | |
| Partition | une tranche d'une table, tenue par un seul travailleur | |
| Job | le travail que Spark fait pour une action |
Les cellules que les étudiants remplissent restent vides dans cet aperçu.
Un DataFrame est une table que l'on interroge
MapReduce vous fait écrire les étapes map et reduce. Spark vous laisse décrire le résultat sur un DataFrame : une table aux colonnes nommées.
Vous interrogez un DataFrame avec du code, comme filter et groupBy, ou avec du SQL. Les deux donnent le même plan. Le labo a chargé un fichier de 24 commandes sous le nom orders.
Comment le labo répond
Le terminal du labo n'a pas Spark. Il envoie vos questions par curl à un petit service, qui répond toujours par un tableau de texte.
Les noms de colonnes viennent du fichier, donc en anglais : country, product, quantity. Ce service affiche même le résultat d'une simple transformation, ce qui force Spark à travailler.
Regarder quatre lignes
curl -s -d 'SELECT * FROM orders LIMIT 4' 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.
Compter les pays différents
curl -s -d 'orders.select("country").distinct().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.
Les transformations attendent, les actions lancent
Une transformation comme filter ou groupBy fabrique un nouveau DataFrame, mais ne lit rien encore. Spark ajoute seulement l'étape à son plan. On dit qu'il est paresseux.
Une action comme count ou collect demande une réponse. Spark exécute alors tout le plan : c'est un job.
Exemple : des étapes et leur sorte
| Étape | Sorte | Lit-elle les données ? |
|---|---|---|
| filter(orders.quantity > 3) | Transformation | Non, elle s'ajoute au plan |
| select("country") | Transformation | Non |
| groupBy("country").sum("quantity") | Transformation | Non |
| count() | Action | Oui, elle exécute le plan |
| collect() | Action | Oui, et ramène toutes les lignes |
Les cellules que les étudiants remplissent restent vides dans cet aperçu.
Un plan qui s'exécute à la demande
# Un plan est une liste d'étapes. Ajouter une étape ne lit aucune donnée.
lignes = [["France", 2], ["Japon", 1], ["France", 5]]
plan = []
plan.append(["garder_plus_de", 1])
print("Plan :", plan)
def executer(lignes, plan):
# Une action exécute tout le plan sur les lignes.
for etape in plan:
if etape[0] == "garder_plus_de":
gardees = []
for ligne in lignes:
if ligne[1] > etape[1]:
gardees.append(ligne)
lignes = gardees
return lignes
print("Résultat :", executer(lignes, plan))
# Remplacez le 1 du plan par 2 puis relancez.
Exécuter un plan : les grosses commandes
curl -s -d 'orders.filter(orders.quantity >= 5).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.
Une erreur fréquente
Attention
On croit souvent qu'une ligne filter lit les données. Elle ne lit rien : elle ajoute une étape au plan. Spark vérifie tout de suite les noms de colonnes, donc une colonne inconnue fait échouer la ligne. Une valeur fausse dans les données, elle, n'apparaît que quand une action exécute le plan.
Les partitions partagent le travail
Spark découpe un DataFrame en partitions. Chaque travailleur additionne sa propre partition. Spark additionne ensuite les sommes partielles.
C'est encore l'idée de map et reduce, sans que vous l'écriviez. Le Spark UI montre chaque job et son plan : vous voyez ce qui a tourné.
Des sommes partielles, puis un total
# Trois partitions de la colonne des quantités, chacune chez un autre travailleur.
partitions = [[3, 1, 2], [4, 2], [6, 1]]
partielles = []
for partie in partitions:
total = 0
for quantite in partie:
total = total + quantite
partielles.append(total)
print("Sommes partielles :", partielles)
general = 0
for valeur in partielles:
general = general + valeur
print("Total :", general)
# Déplacez le 6 dans la première partition puis relancez.
Demander en combien de partitions est la table
curl -s -d 'orders.rdd.getNumPartitions()' 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.
Couper la table en trois tranches
curl -s -d 'orders.repartition(3).rdd.glom().map(len).collect()' 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.
Commandes reçues pour chaque pays
curl -s -d 'SELECT country, count(*) AS commandes FROM orders GROUP BY country ORDER BY commandes DESC, 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.
Le Spark UI montre chaque job
Ouvrez Spark UI depuis la barre du labo. L'onglet Jobs liste le travail fait pour chaque action. L'onglet SQL / DataFrame liste chaque requête avec son plan.
Relancez une commande ci-dessus et regardez une nouvelle ligne apparaître. C'est ainsi qu'un ingénieur comprend pourquoi un pipeline est lent.
Pour aller plus loin : votre propre requête
Écrivez votre propre requête SQL sur orders. Pour chaque produit, elle donne le nombre de commandes et la quantité moyenne par commande. Triez de la plus grande moyenne à la plus petite.
Avant de l'exécuter, prévoyez quel produit arrive en tête. Exécutez-la, puis retrouvez son job dans l'onglet SQL / DataFrame du Spark UI.