Introduction
Une petite galerie permet aux visiteurs de voter pour leur stand préféré. Un émetteur peut réessayer un événement ; un nouvel enregistrement Kafka ne signifie donc pas toujours un vote supplémentaire. Vous observerez d’abord un exemple normal à deux votes, puis une véritable nouvelle tentative stockée à une autre position, avant de reconstruire les décomptes de la galerie.
Vous devez déjà connaître les chemins du terminal, les fonctions/boucles/dictionnaires Python, JSON, ainsi que les bases des clés, partitions, offsets, groupes de consommateurs et du client Python Kafka présentées dans Apache Kafka for Beginners. Ce laboratoire ajoute l’identité d’un événement métier. Un lecteur natif complet est fourni pour vous permettre de vous concentrer sur la règle de décompte ; le laboratoire guidé suivant construit ce lecteur. Utilisez Terminal et l’environnement bigdata-python installé. L’estimation de 40–50 minutes de l’auteur comprend la lecture et les erreurs courantes ; elle ne provient pas d’un essai chronométré auprès de débutants.
Il s’agit uniquement d’un lot fini déjà livré. Ce laboratoire ne met pas en place de contrôle d’éligibilité des votants ni de flux continu, et ne promet ni reprise après plantage ni traitement exactement une fois de bout en bout. Chaque laboratoire démarre indépendamment.
Observez une nouvelle tentative à une autre position stockée
Dans cette étape, vous comparerez deux votes ordinaires à un événement réémis. Commencez dans Terminal. L’espace de travail contient des outils et des catalogues préparés. Modifiez uniquement reducer.py, replay.py et, si nécessaire, des auxiliaires Python sous helpers/ ; placez les fichiers dérivés sous output/. Préservez les outils/données fournis, les topics sources figés et les groupes des applications d’affichage suspendues. Le topic de démonstration distinct autorise une seule action : la nouvelle tentative ci-dessous.
cd /home/labex/project
-B empêche ces commandes Python de créer des fichiers de cache de bytecode. L’outil en lecture seule inspect affiche l’enveloppe réelle de chaque enregistrement du broker : topic, partition, offset, clé et valeur décodée. Son total final compte les enregistrements applicatifs stockés. Inspectez la petite démonstration :
bigdata-python -B vote_tool.py inspect demo
Deux enregistrements sont attendus. Leurs valeurs respectent ce schéma : event_id, booth et choice sont des chaînes non vides ; choice vaut exactement favourite ; booth doit figurer dans le catalogue sélectionné. Aucun champ supplémentaire n’appartient au schéma de vote de ce cours. La clé est le nom du stand, utilisé pour le routage. event_id identifie un vote métier. Ici, demo-1 vote pour oak et demo-2 pour fern : deux enregistrements et deux votes logiques.
Avant l’envoi, prévoyez les deux résultats : combien d’enregistrements physiques seront stockés, et quels seront les décomptes des stands si demo-1 reste un seul vote métier ? Testez ensuite cette prévision.
La commande de nouvelle tentative envoie la même valeur complète de demo-1 au véritable broker et attend la confirmation de livraison. Elle vérifie d’abord la démonstration existante : l’exécuter de nouveau une fois la tentative attendue présente n’ajoute donc pas de quatrième enregistrement :
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
Il y a maintenant trois enregistrements. demo-1 apparaît à deux offsets différents avec une valeur au contenu identique. Ces positions identifient deux enregistrements stockés, tandis que l’ID d’événement répété identifie toujours un seul vote. Les totaux corrects restent oak=1, fern=1, reed=0. Relire un enregistrement existant et stocker une nouvelle tentative sont deux situations différentes ; toutes deux nécessitent une règle d’identité métier pour établir les décomptes.
Le lot figé du cours utilise un placement explicite dans les partitions ; des clés de stand identiques peuvent donc apparaître dans des partitions différentes. Consultez les métadonnées réelles lorsque vous comparez les enregistrements.
Dans le lot sélectionné du cours, un ID d’événement est global à toutes les partitions, et non limité à une partition ou à un stand. Deux ID différents comptent comme deux votes même si tous les autres champs sont identiques. Le métier fourni accepte ces événements ; il ne décide pas combien de votes une personne réelle peut émettre.
Si un outil signale un lot inconnu, affichez les noms préparés et utilisez demo pour cette observation :
bigdata-python -B vote_tool.py list
Une faute de frappe n’a pas réinitialisé le broker. Corrigez l’argument et répétez l’inspection en lecture seule. Ne renvoyez pas d’événements arbitraires pour corriger un décompte affiché.
Conservez les identités logiques et refusez les contenus contradictoires
Dans cette étape, vous implémenterez la règle métier dans la fonction Python préparée. Restez dans Terminal, à /home/labex/project. Lisez le fichier actuel :
cat reducer.py
Il initialise les décomptes du catalogue, puis incrémente pour chaque valeur d’entrée. Prévoyez son résultat avec les trois enregistrements de démonstration : le deuxième demo-1 stocké ajouterait un vote. La règle correcte doit compter cet ID une seule fois tout en gardant les ID distincts séparés.
Le vérificateur préparé appelle une copie exacte de votre fonction sur des exemples jetables, notamment des valeurs réordonnées et d’autres noms de stands. Il ne modifie pas le journal Kafka actif. Exécutez la fonction naïve actuelle ; cette commande doit échouer avec un message concernant la règle :
bigdata-python -B check_rules.py
Le shell conserve dans $? le code de sortie de la commande immédiatement précédente : zéro signifie réussite, une valeur non nulle signifie échec. Affichez-le immédiatement, avant qu’une autre commande ne le remplace :
printf '%s\n' "$?"
summarize(values, booths) doit renvoyer un dictionnaire contenant votes et counts pour une entrée valide, et lever une exception ordinaire pour une entrée invalide. Gardez le module appelable ; n’y terminez pas l’interpréteur Python. Le vérificateur doit recevoir les valeurs réellement renvoyées, pas seulement un statut de sortie de l’interpréteur.
Modifiez la fonction existante dans nano
Ouvrez le fichier modifiable dans l’éditeur de texte du terminal :
nano reducer.py
La barre supérieure indique le nom du fichier. Déplacez-vous avec les flèches et corrigez les caractères avec Retour arrière/Suppr. Python utilise l’indentation pour regrouper les instructions : utilisez quatre espaces dans la fonction, huit dans sa boucle et quatre de plus pour chaque condition imbriquée. Gardez la ligne existante def summarize(values, booths):. Supprimez l’ancien commentaire sur le comptage volontaire des enregistrements lorsque vous remplacez cette logique.
Les courts blocs Python suivants sont des modifications liées dans cette unique fonction, pas des commandes à exécuter dans le shell ni des programmes séparés. Terminez toutes les modifications avant d’exécuter le vérificateur. Gardez les parties valides du code de départ plutôt que de remplacer tout le fichier.
D’abord, gardez l’initialisation existante des décomptes à zéro et ajoutez seen immédiatement après, avant la boucle. Un dictionnaire indexé par event_id contiendra la valeur complète de chaque vote métier accepté :
counts = {booth: 0 for booth in booths}
seen = {}
Ensuite, gardez l’unique ligne existante for value in values: et remplacez son incrémentation initiale par la validation ci-dessous. Le bloc montre cette ligne for existante pour situer la modification ; n’ajoutez pas de deuxième boucle. set(value) rassemble les noms de champs ; isinstance vérifie les types ; any(...) demande si au moins une valeur de champ est invalide. La première condition exige un dictionnaire avec exactement trois champs de chaîne non vide et le choix autorisé. Vérifiez le catalogue avant de tenter d’incrémenter un stand. Gardez la ligne for à quatre espaces et sa validation à huit :
for value in values:
if (not isinstance(value, dict)
or set(value) != {"event_id", "booth", "choice"}
or any(not isinstance(item, str) or not item for item in value.values())
or value["choice"] != "favourite"):
raise ValueError("Invalid vote schema")
if value["booth"] not in counts:
raise ValueError("Unknown booth: " + value["booth"])
Ajoutez maintenant le traitement de l’identité immédiatement après cette vérification du catalogue, toujours dans la boucle. Pour un ID connu, comparez la valeur décodée complète. Un contenu différent doit lever une erreur nommant cet ID ; choisir la première ou la dernière valeur rendrait la réponse dépendante de l’ordre de parcours. Une tentative identique atteint continue, qui passe à l’entrée suivante sans nouvelle incrémentation. Un premier ID est enregistré et compté :
event_id = value["event_id"]
if event_id in seen:
if seen[event_id] != value:
raise ValueError("Conflicting event_id: " + event_id)
continue
seen[event_id] = dict(value)
counts[value["booth"]] += 1
dict(value) conserve la valeur acceptée sous forme de dictionnaire. Les espaces JSON et l’ordre des champs ne définissent pas l’identité métier. La clé du dictionnaire est globale aux partitions du lot sélectionné ; ce n’est ni un nom de stand ni un couple partition/offset. Des ID distincts avec le même stand et le même choix doivent chacun atteindre l’incrémentation des premiers ID.
Enfin, remplacez l’ancien retour qui utilisait toutes les valeurs d’entrée. Renvoyez les valeurs uniques acceptées et les décomptes de tous les stands du catalogue. Cette ligne a quatre espaces, hors de la boucle mais dans la fonction :
return {"votes": list(seen.values()), "counts": counts}
Enregistrez et quittez. Dans les libellés du bas de nano, ^ signifie maintenir Ctrl : appuyez sur Ctrl+O, puis sur Entrée pour confirmer reducer.py ; appuyez sur Ctrl+X pour revenir au shell. Ces touches enregistrent le fichier courant, sans exécuter Python. Relisez ce que vous avez réellement enregistré :
cat reducer.py
Vérifiez que seen est initialisé une fois par appel de fonction, que les contrôles du schéma et du catalogue précèdent le comptage, que seul un premier ID incrémente et que le retour est hors de la boucle. Testez ensuite immédiatement la fonction terminée :
bigdata-python -B check_rules.py
Vous devez voir Rules pass pour les identités, les ID distincts, le conflit, le schéma, les stands sans vote et les entrées réordonnées. Un dictionnaire indexé uniquement par stand fusionnerait des événements de vote distincts ; un dictionnaire indexé par partition/offset conserverait toutes les nouvelles tentatives. Votre dictionnaire représente l’identité métier.
Pour une erreur d’indentation, de syntaxe ou de nom, lisez le fichier et la ligne signalés, relancez nano reducer.py et corrigez la partie concernée avec les flèches. Enregistrez avec Ctrl+O, Entrée, quittez avec Ctrl+X, inspectez avec cat reducer.py et relancez la vérification de la règle. Gardez les parties déjà correctes ; vous n’avez pas besoin de réécrire toute la fonction ni de réinitialiser la VM. Ces commandes en lecture seule permettent de retrouver un mauvais chemin :
pwd
ls reducer.py replay.py
Revenez avec cd /home/labex/project si nécessaire. L’étape suivante utilise le lecteur natif préparé pour appliquer votre règle à de véritables enregistrements Kafka.
Publiez et lisez un instantané du véritable journal de la galerie
Dans cette étape, vous appliquerez votre fonction au véritable lot figé de la galerie. Restez dans Terminal, à /home/labex/project. Affichez la source et la sortie choisies par le lot main, puis examinez son catalogue :
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main désigne gallery-votes. Il comporte plusieurs partitions, des ID répétés à des positions stockées réellement différentes, des ID distincts dont le reste du contenu est égal et une partition vide. replay.py est le lecteur natif préparé ; vote_support.py fournit l’analyse des arguments et l’écriture de l’instantané. Le lecteur transmet la valeur de chaque enveloppe réelle à votre fonction summarize. Il ne récupère pas les valeurs depuis un fichier local de votes d’entrée.
Exécutez le point d’entrée public ci-dessous. Il vérifie d’abord la fonction d’identité actuelle sur des exemples jetables, puis exécute votre programme actuel et ses processus enfants auxiliaires, conserve des preuves de lecture native pour cet arbre de processus et enregistre le résultat réussi dans le chemin de sortie indiqué :
bigdata-python -B run_replay.py main
Une publication avec records=8, logical_votes=5 est attendue. Le point d’entrée enregistre une véritable réception des données du broker par cet arbre de programmes, au lieu de se fier à un consommateur sans rapport qui utilise la même étiquette client. Il ne démontre ni la causalité du calcul interne ni une protection contre la falsification par un administrateur. Le résultat métier final est vérifié séparément par rapport à la source réelle complète.
Chaque exécution réussie doit réellement publier son résultat pendant cet appel : écrire directement les données du fichier, ou écrire un fichier temporaire puis le renommer en résultat. Un processus enfant auxiliaire peut effectuer l’écriture. Republier des octets identiques est valide ; lire Kafka tout en conservant un ancien fichier n’est pas une nouvelle publication.
Lisez le véritable fichier de résultat. Ce lecteur ne récupère, ne complète et ne répare jamais silencieusement un résultat manquant :
bigdata-python -B read_snapshot.py output/main-snapshot.json
Cinq ID de vote logique et les décomptes suivants sont attendus :
Booth Votes
fern 2
oak 3
reed 0
L’instantané conserve aussi les huit enveloppes physiques ; vous pouvez ainsi distinguer ses enregistrements sources de ses cinq votes logiques. reed reste présent avec zéro vote. Avec le mécanisme d’écriture préparé, une erreur de boucle ou de validation avant publication laisse l’instantané précédent disponible. Le point d’entrée n’annule pas automatiquement les autres méthodes d’écriture. Après toute modification du programme, relancez le point d’entrée public avant de vérifier la réussite.
L’application d’affichage originale est suspendue et conserve la progression de son groupe de consommateurs. Vous pouvez décrire ce groupe, mais vous ne devez ni le rejoindre, ni le réinitialiser/supprimer, ni y valider des offsets. Ne modifiez/supprimez pas le topic source et n’ajoutez pas de partitions. Reconstruisez uniquement votre sortie dérivée. La vérification observe l’identité du topic d’origine, la topologie complète des partitions/répliques, les enregistrements réels et la progression enregistrée de l’application ; elle ne promet pas de préserver toute configuration arbitraire du broker.
Vous disposez maintenant d’un instantané utile des stands, et pas seulement d’une trace de livraison. Le laboratoire suivant explique et construit le lecteur natif indépendant complet, notamment pourquoi un sondage vide ne signifie pas que le lot est terminé.
Résumé
Vous avez observé une véritable nouvelle tentative stockée, distingué l’identité métier globale des positions d’enregistrement Kafka et refusé des contenus contradictoires portant le même ID. Votre instantané réel de la galerie conserve toute la source physique tout en présentant les votes logiques et chaque stand, y compris ceux sans vote. La source et le groupe de l’application d’affichage suspendue restent disponibles.
