Introducción
Una pequeña galería permite a los visitantes votar por su puesto favorito. El emisor puede reintentar un evento, por lo que un nuevo registro de Kafka no siempre significa otro voto. Primero observarás una muestra normal de dos votos, después un reintento real en otra posición almacenada y, por último, reconstruirás los recuentos de la galería.
Debes conocer las rutas de terminal, las funciones/bucles/diccionarios de Python, JSON y los fundamentos de claves, particiones, offsets, grupos de consumidores y cliente Python de Kafka tratados en Apache Kafka for Beginners. Este laboratorio añade la identidad del evento de negocio. Se proporciona un lector nativo completo para que te centres en la regla de recuento; el siguiente laboratorio guiado construye ese lector. Usa Terminal y el entorno bigdata-python instalado. La estimación del autor de 40–50 minutos incluye lectura y errores habituales; no es una medición con principiantes.
Solo se trabaja con un lote finito ya entregado. No se implementa un sistema de elegibilidad de votantes, un flujo continuo, una garantía de recuperación ante caídas ni procesamiento exactamente una vez de extremo a extremo. Cada laboratorio empieza de forma independiente.
Observa un reintento en otra posición almacenada
En este paso, compararás dos votos normales con un evento reintentado. Empieza en Terminal. El espacio de trabajo contiene herramientas y catálogos preparados. Edita únicamente reducer.py, replay.py y, si los necesitas, auxiliares Python bajo helpers/; guarda los archivos derivados bajo output/. Conserva las herramientas/datos suministrados, los temas de origen congelados y los grupos de las aplicaciones de visualización pausadas. El tema de demostración separado solo permite una acción: el reintento que se muestra a continuación.
cd /home/labex/project
-B evita que estos comandos Python creen archivos de caché de bytecode. La herramienta de solo lectura inspect muestra la envoltura real de cada registro del broker: tema, partición, offset, clave y valor decodificado. Su recuento final indica cuántos registros de aplicación se han almacenado. Inspecciona la pequeña demostración:
bigdata-python -B vote_tool.py inspect demo
Debes ver dos registros. Sus valores tienen este esquema: event_id, booth y choice son cadenas no vacías; choice es exactamente favourite; booth debe aparecer en el catálogo seleccionado. El esquema de voto de este curso no admite campos adicionales. La clave es el nombre del puesto, utilizado para el enrutamiento. event_id identifica un voto de negocio. Aquí, demo-1 vota por oak y demo-2 por fern: dos registros y dos votos lógicos.
Antes de enviar, predice ambos resultados: ¿cuántos registros físicos se almacenarán y cuáles serán los recuentos por puesto si demo-1 sigue siendo un único voto de negocio? Después comprueba esa predicción.
El comando de reintento envía el mismo valor completo de demo-1 al broker real y espera la confirmación de entrega. Antes de enviarlo comprueba la demostración existente, por lo que volver a ejecutarlo tras el reintento esperado no añade un cuarto registro:
bigdata-python -B vote_tool.py retry-demo
bigdata-python -B vote_tool.py inspect demo
Ahora hay tres registros. demo-1 aparece en dos offsets distintos con un contenido de valor idéntico. Esas posiciones identifican dos registros almacenados, mientras que el ID de evento repetido sigue identificando un solo voto. Los totales correctos siguen siendo oak=1, fern=1, reed=0. Leer de nuevo un registro existente y almacenar otro reintento son situaciones diferentes; ambas necesitan una regla de identidad de negocio para construir los recuentos.
El lote congelado del curso usa colocación explícita en particiones, por lo que claves de puesto iguales pueden aparecer en particiones diferentes. Lee los metadatos reales al comparar registros.
Dentro del lote seleccionado del curso, el ID de evento es global en todas las particiones, no solo dentro de una partición o un puesto. Dos ID distintos cuentan como dos votos aunque todos los demás campos sean iguales. El negocio suministrado acepta esos eventos; no determina cuántos votos puede emitir una persona real.
Si una herramienta indica que el lote es desconocido, enumera los nombres preparados y usa demo para esta observación:
bigdata-python -B vote_tool.py list
Un error de escritura no ha restablecido el broker. Corrige el argumento y repite la inspección de solo lectura. No vuelvas a enviar eventos arbitrarios para corregir un recuento mostrado.
Conserva las identidades lógicas y rechaza el contenido conflictivo
En este paso, implementarás la regla de negocio dentro de la función Python preparada. Continúa en Terminal, en /home/labex/project. Lee el archivo actual:
cat reducer.py
Inicializa los recuentos del catálogo y después incrementa por cada valor de entrada. Predice qué haría con los tres registros de demostración: el segundo demo-1 almacenado añadiría otro voto. La regla correcta debe contar ese ID una sola vez, manteniendo separados los ID distintos.
El comprobador de reglas preparado llama a una copia exacta de tu función con ejemplos desechables, incluidos valores reordenados y nombres de puesto diferentes. No cambia el log activo de Kafka. Ejecuta la función ingenua actual; se espera que este comando falle con un mensaje sobre la regla:
bigdata-python -B check_rules.py
El shell guarda el estado de salida del comando inmediatamente anterior en $?: cero significa éxito y un valor distinto de cero significa fallo. Muéstralo inmediatamente, antes de que otro comando lo sustituya:
printf '%s\n' "$?"
summarize(values, booths) debe devolver un diccionario que contenga votes y counts para entradas válidas y lanzar una excepción ordinaria para entradas inválidas. Mantén el módulo invocable; no termines el intérprete de Python desde él. El comprobador debe recibir valores devueltos reales, no solo un estado de salida del intérprete.
Edita la función existente en nano
Abre el archivo editable en el editor de texto de Terminal:
nano reducer.py
La barra superior identifica el nombre de archivo. Usa las flechas para moverte y Backspace/Delete para corregir caracteres. Python usa la sangría para agrupar instrucciones: usa cuatro espacios dentro de la función, ocho dentro de su bucle y otros cuatro por cada condición anidada. Conserva la línea existente def summarize(values, booths):. Elimina el comentario antiguo sobre contar registros intencionadamente cuando sustituyas esa lógica.
Los pequeños bloques Python siguientes son ediciones conectadas dentro de esta misma función, no comandos para ejecutar en el shell ni programas separados. Completa todas las ediciones antes de ejecutar el comprobador. Conserva las partes válidas del código inicial en lugar de sustituir todo el archivo.
Primero conserva la inicialización existente de los recuentos a cero y añade seen justo debajo, antes del bucle. Un diccionario indexado por event_id guardará el valor completo de cada voto de negocio aceptado:
counts = {booth: 0 for booth in booths}
seen = {}
Después conserva la única línea existente for value in values: y sustituye su incremento original por la validación siguiente. El bloque muestra esa línea for existente como contexto; no añadas un segundo bucle. set(value) reúne los nombres de campo; isinstance comprueba tipos; any(...) pregunta si al menos un valor de campo es inválido. La primera condición exige un diccionario con exactamente tres campos de cadena no vacía y la elección aprobada. Comprueba el catálogo antes de intentar incrementar un puesto. Mantén la línea for con cuatro espacios y su validación con ocho:
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"])
Añade ahora el tratamiento de identidad inmediatamente después de esa comprobación del catálogo, todavía dentro del bucle. Ante un ID conocido, compara el valor decodificado completo. Un contenido distinto debe lanzar un error que nombre ese ID; elegir el primer o último valor haría que la respuesta dependiera del orden de recorrido. Un reintento idéntico llega a continue, que pasa a la siguiente entrada sin otro incremento. Un ID nuevo se guarda y se cuenta:
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) conserva el valor aceptado como diccionario. Los espacios y el orden de campos de JSON no definen la identidad de negocio. La clave del diccionario es global en las particiones del lote seleccionado; no es un nombre de puesto ni un par partición/offset. Los ID distintos con el mismo puesto y elección deben llegar cada uno al incremento de un ID nuevo.
Por último, sustituye el retorno antiguo que usaba todos los valores de entrada. Devuelve los valores únicos aceptados y todos los recuentos del catálogo. Esta línea tiene cuatro espacios, fuera del bucle pero dentro de la función:
return {"votes": list(seen.values()), "counts": counts}
Guarda y sal. En las etiquetas inferiores de nano, ^ significa mantener Ctrl: pulsa Ctrl+O y después Enter para confirmar reducer.py; pulsa Ctrl+X para volver al shell. Estas teclas guardan el archivo actual, no ejecutan Python. Lee lo que realmente guardaste:
cat reducer.py
Comprueba que seen se inicialice una vez por llamada de función, que las comprobaciones de esquema/catálogo precedan al recuento, que solo el primer ID incremente y que el retorno esté fuera del bucle. Después prueba inmediatamente la función completada:
bigdata-python -B check_rules.py
Debes ver Rules pass para identidades, ID distintos, conflicto, esquema, puestos con cero votos y entrada reordenada. Un diccionario cuya clave fuera solo el puesto fusionaría eventos de voto distintos; uno cuya clave fuera partición/offset conservaría todos los reintentos. Tu diccionario representa la identidad de negocio.
Para un error de sangría, sintaxis o nombre, lee el archivo y la línea indicados, ejecuta nano reducer.py de nuevo y usa las flechas para corregir la parte afectada. Guarda con Ctrl+O, Enter, sal con Ctrl+X, inspecciona con cat reducer.py y repite la comprobación de reglas. Conserva las partes correctas; no necesitas reescribir toda la función ni restablecer la VM. Puedes localizar una ruta incorrecta con estos comandos de solo lectura:
pwd
ls reducer.py replay.py
Vuelve con cd /home/labex/project si es necesario. El siguiente paso usa el lector nativo preparado para aplicar tu regla a registros reales de Kafka.
Publica y lee una instantánea del log real de la galería
En este paso, aplicarás tu función al lote real congelado de la galería. Continúa en Terminal, en /home/labex/project. Enumera el origen y la salida seleccionados por el lote main e inspecciona después su catálogo:
bigdata-python -B vote_tool.py list
cat data/main-catalog.json
main corresponde a gallery-votes. Tiene varias particiones, ID repetidos en posiciones almacenadas realmente distintas, ID diferentes con el resto del contenido igual y una partición vacía. replay.py es el lector nativo preparado; vote_support.py proporciona el análisis de argumentos y la escritura de instantáneas. El lector pasa el valor de cada envoltura real a tu función summarize. No obtiene valores de un archivo local de entrada de votos.
Ejecuta el punto de entrada público siguiente. Primero comprueba la función de identidad actual con ejemplos desechables, luego ejecuta tu programa actual y sus procesos hijos auxiliares, conserva evidencia de lectura nativa de ese árbol de procesos y guarda el resultado correcto en la ruta de salida indicada:
bigdata-python -B run_replay.py main
Debes obtener una publicación con records=8, logical_votes=5. El punto de entrada registra la recepción real de datos del broker por este árbol de programas, en lugar de confiar en un consumidor ajeno con la misma etiqueta de cliente. No demuestra causalidad del cálculo interno ni protección frente a la falsificación por un administrador. El resultado de negocio final se comprueba por separado contra el origen real completo.
Cada ejecución correcta debe publicar realmente el resultado durante esa invocación: escribir datos directamente en el archivo, o escribir un archivo temporal y renombrarlo como resultado. Un proceso hijo auxiliar puede realizar la escritura. Es válido publicar de nuevo bytes idénticos; leer Kafka y conservar un archivo antiguo no constituye una nueva publicación.
Lee el archivo de resultado real. Este lector nunca obtiene, completa ni repara silenciosamente resultados ausentes:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Debes ver cinco ID de voto lógico y estos recuentos por puesto:
Booth Votes
fern 2
oak 3
reed 0
La instantánea también conserva las ocho envolturas físicas, por lo que puedes distinguir los registros de origen de sus cinco votos lógicos. reed sigue presente con cero votos. Con el escritor preparado, los errores de bucle o validación anteriores a la publicación dejan disponible la instantánea previa. El punto de entrada no revierte automáticamente otros métodos de escritura. Tras editar el programa, vuelve a ejecutar el punto de entrada público antes de comprobar la finalización.
La aplicación de visualización original está pausada y conserva el progreso de su grupo de consumidores. Puedes describir ese grupo, pero no unirte a él, restablecerlo/eliminarlo ni confirmar offsets en él. No edites/elimines el tema de origen ni añadas particiones. Reconstruye solo tu propia salida derivada. La comprobación observa la identidad del tema original, la topología completa de particiones/réplicas, los registros reales y el progreso guardado de la aplicación de visualización; no promete preservar cualquier configuración del broker.
Ahora tienes una instantánea útil de los puestos, no solo una transcripción de entregas. El siguiente laboratorio explica y construye el lector nativo independiente completo, incluido por qué un sondeo sin mensajes no significa que el lote haya terminado.
Resumen
Has observado un reintento almacenado real, separado la identidad global de negocio de las posiciones de registro de Kafka y rechazado contenido conflictivo con el mismo ID. Tu instantánea real de la galería conserva el origen físico completo y muestra los votos lógicos y todos los puestos, incluidos los que tienen cero votos. El origen y el grupo de la aplicación de visualización pausada siguen disponibles.
