Introducción
Un estudio ha conservado su historial de votos mientras su aplicación de visualización original está pausada. Necesitas un recuento nuevo sin rebobinar esa aplicación ni cambiar el log. Un sondeo puede no devolver mensajes durante un instante aunque un consumidor nuevo todavía no haya recibido el lote congelado.
Construirás el bucle de relectura nativa, capturarás la posición final de cada partición por separado y publicarás solo después de leer el lote finito completo. La función de identidad lógica del laboratorio anterior ya está disponible en esta VM independiente. Necesitas los fundamentos de Python y Kafka utilizados allí. Usa Terminal. La estimación del autor de 50–60 minutos incluye lectura y errores habituales; no es una prueba cronometrada con principiantes.
Se trata de reconstruir un lote congelado. No se sigue publicando en estos temas durante el ejercicio, no hay que implementar transacciones y no se garantiza recuperación fiable ante caídas ni procesamiento exactamente una vez de extremo a extremo.
Captura el límite de cada partición y lee hasta alcanzarlo
En este paso, implementarás una lectura nativa independiente completa dentro de un programa exterior preparado. Empieza en Terminal. Edita replay.py y, si hace falta, reducer.py o auxiliares Python bajo helpers/; escribe los archivos derivados bajo output/. Conserva las herramientas/datos suministrados, los temas originales y los grupos de las aplicaciones de visualización pausadas:
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
La descripción de solo lectura muestra los ID de partición reales de studio-votes, el inicio retenido low, el final congelado actual y las réplicas. Una partición es un log ordenado, no una VM separada. El final es la posición siguiente al lote retenido, no un registro que debas contar. Predice qué particiones necesitan registros: una partición vacía tiene low == high y ya está completa.
El intervalo de offsets no es un número de registros. Kafka puede saltar posiciones internas/de control o eliminadas. Este origen incluye una posición nativa omitida; no hay un voto ausente que debas inventar. Inspecciona las envolturas reales y compara sus posiciones efectivas con la descripción:
bigdata-python -B vote_tool.py inspect main
Antes de reconstruir, lee las posiciones guardadas de la aplicación de visualización pausada con la herramienta de grupos de consumidores del curso previo:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Conserva cada CURRENT-OFFSET para compararlo después. Es normal que no haya miembros activos: la aplicación está pausada, pero tiene progreso guardado. Aquí, LAG es una diferencia entre posiciones del log, no un recuento de votos lógicos. Tu consumidor independiente no debe alterar el progreso de este grupo.
Lee el programa exterior preparado
cat replay.py
Las importaciones, arguments(), la configuración del cliente, el informe de errores y finally/cierre ya están suministrados. El núcleo lanza deliberadamente NotImplementedError; no lee particiones ni publica una respuesta de ejemplo. La función de identidad correcta del laboratorio anterior está preparada por separado en reducer.py.
El cliente usa la dirección bootstrap.servers seleccionada y un group.id independiente. allow.auto.create.topics es falso. La confirmación automática y el almacenamiento automático de offsets están desactivados. assign le dará posiciones iniciales explícitas sin asignación dinámica de grupos; este ejemplo no realiza llamadas explícitas de confirmación. Una implementación equivalente con un grupo independiente puede guardar su propio progreso, pero nunca debe unirse a un grupo de visualización original, confirmar en él, restablecerlo ni eliminarlo.
El vote_support.py preparado suministra arguments() para tema/catálogo/salida/broker/grupo y finish(...) para escribir el resultado completo. No descubre particiones ni obtiene votos. finish llama a la regla de identidad antes de escribir/sustituir el resultado. finally cierra el consumidor tanto si hay éxito como si hay fallo; conserva esa estructura exterior.
Implementa el núcleo nativo en partes cortas conectadas
Usa el flujo del editor del laboratorio anterior:
nano replay.py
Conserva las importaciones, el análisis de argumentos, la configuración de Consumer y las secciones try:, except y finally. Dentro de try:, elimina únicamente los dos comentarios explicativos y el raise NotImplementedError(...) del núcleo incompleto. Usa las flechas para situarte y Backspace/Delete para editar; Ctrl+K puede cortar toda la línea actual, así que úsalo solo cuando estés en una línea que quieras sustituir.
Inserta en ese hueco las partes conectadas siguientes. Forman un único núcleo: no ejecutes bloques individuales en el shell ni guardes/ejecutes un bucle a medio construir. Completa todas las partes, conserva el marco exterior de errores/cierre suministrado y después guarda y ejecuta. Cuatro espacios sitúan una instrucción dentro de try:; otros cuatro la sitúan dentro de un bucle o condición. La sangría relativa mostrada forma parte del significado de Python.
Primero descubre el conjunto completo de particiones reales. Llamar a list_topics sin un tema solicitado evita pedir metadatos de un nombre desconocido. Comprueba la disponibilidad antes de usar la entrada seleccionada. Ordenar los ID descubiertos da a este ejemplo una lista de asignación estable; no crea un orden global de eventos:
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
Para cada partición descubierta, solicita get_watermark_offsets(..., timeout=10) una vez. TopicPartition identifica el tema y la partición de esa solicitud. Conserva tanto el inicio retenido low como el final congelado high e inicializa la siguiente posición diagnóstica en low. No actualices high dentro del bucle de sondeo: perseguir un final creciente dejaría de ser un recuento finito. Este ejercicio prohíbe añadir o modificar el origen congelado:
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
Construye pending solo con las particiones que tengan posiciones retenidas por leer. assign recibe una lista de offsets iniciales explícitos, incluidos los lows de las particiones vacías. Inicializa una lista de registros físicos nueva para cada ejecución. time.monotonic() mide el tiempo transcurrido independientemente de ajustes del reloj del sistema; el plazo limita una lectura incompleta:
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
Inicia ahora el bucle. poll(0.2) puede devolver un mensaje, un mensaje con un error/evento o None al agotarse la espera. Predice por qué None no puede significar que haya terminado: puede que el cliente todavía no haya recibido un registro congelado. Sigue esperando dentro del plazo total. Los eventos EOF de partición están activados en la configuración; _PARTITION_EOF es el evento de final esperado, mientras que otros errores hacen fallar la ejecución. Esto permite observar el avance nativo más allá de una posición interna final:
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
Continúa el mismo bloque if message is not None con el elif siguiente, alineado con if message.error(): a doce espacios. Solo un mensaje normal de una partición pendiente, por debajo de su high fijo, se convierte en una envoltura. Conserva también los reintentos físicos; summarize trata después la identidad lógica. Decodifica los bytes de clave/valor en UTF-8; después, json.loads convierte el valor en un diccionario:
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
Después de tratar este sondeo, consulta consumer.position(...) para cada partición restante. Es la siguiente posición del consumidor, distinta del progreso confirmado de cualquier grupo. Compara cada una con su high capturado, no con un número supuesto de registros. Elimina de pending únicamente las particiones completadas. La línea for position tiene ocho espacios, dentro de while pero fuera de la condición del mensaje, de modo que incluso un sondeo inactivo puede ir seguido de una comprobación de posición. Por último, llama a finish con cuatro espacios, después de que haya terminado el bucle:
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
Cada partición tiene ahora un límite fijo. Las vacías no necesitaron votos; las posiciones omitidas no exigieron inventar registros; None no detuvo la lectura. Solo cuando todas las siguientes posiciones pendientes alcanzan sus highs se publica una instantánea completa.
Guarda con Ctrl+O, Enter para confirmar replay.py y después Ctrl+X. Inspecciona el programa guardado:
cat replay.py
Comprueba que las seis partes estén dentro del try suministrado, que el tratamiento de envolturas permanezca en la rama de mensajes normales, que la comprobación de posiciones esté dentro del bucle y que finish esté fuera. Ejecuta inmediatamente contra el lote nativo seleccionado:
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
Debes obtener ocho registros físicos, cinco votos lógicos y cedar=3, maple=2, willow=0. El punto de entrada público primero comprueba la función de identidad con ejemplos desechables y después registra datos del broker recibidos por el árbol nativo real del programa/auxiliares. Una comparación completa de negocio separada comprueba el resultado real. Los límites/posiciones declarados por la instantánea son diagnósticos, no una prueba independiente de procesamiento nativo.
Cada invocación correcta debe escribir realmente su resultado, directamente o mediante un archivo temporal escrito y renombrado como salida seleccionada. Se aceptan bytes idénticos al repetir, y los procesos hijos auxiliares pueden escribir. Conservar un archivo antiguo o actualizar solo su marca de tiempo, o escribir un marcador de finalización, no constituye una publicación. El finish suministrado utiliza el método enseñado de archivo temporal/sustitución. Con ese 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 escrituras arbitrarias directamente sobre el archivo de destino.
Para un error de sintaxis/nombre/sangría, usa la línea indicada para volver a abrir nano replay.py, corrige esa parte y conserva las piezas que funcionan. Guarda y sal, inspecciona con cat replay.py y vuelve a ejecutar el punto de entrada público. Para un tiempo de espera por relectura incompleta, inspecciona el argumento del tema seleccionado y el tratamiento de pending, EOF y posiciones; no es motivo para rebobinar el grupo original. Usa pwd y ls replay.py reducer.py para localizar una ruta mal escrita. Un error local del programa no requiere volver a ejecutar la preparación ni cambiar registros de origen.
Reconstruye desde cero y utiliza otro lote
En este paso, demostrarás que otra ejecución reconstruye el mismo resultado finito y luego usarás un catálogo y una distribución de particiones nuevos. Continúa en Terminal, en /home/labex/project.
El escritor preparado ordena las envolturas físicas por partición/offset y los votos lógicos por ID antes de guardar JSON. Así, la misma instantánea completa es estable sin afirmar que exista un orden global de eventos. sha256sum calcula una huella del contenido. > guarda esa huella bajo el directorio de salida; después, -c compara el archivo real indicado con la huella guardada:
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
Debes ver output/main-snapshot.json: OK. En cada invocación, el programa empieza con records, seen y recuentos vacíos. Sustituye una instantánea completa, en lugar de volver a sumar los votos históricos a los totales anteriores. El consumo no eliminó esos registros ni rebobinó la aplicación de visualización pausada.
Vuelve a leer las posiciones de la aplicación original después de repetir la ejecución:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Todos los valores CURRENT-OFFSET deben coincidir con la tabla anterior de la aplicación pausada. Tu reconstrucción independiente no confirmó offsets en ese grupo ni lo rebobinó, aunque leyó los registros retenidos. Sigue siendo normal que no haya miembros activos.
Enumera el lote check e inspecciona su propio catálogo de puestos:
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
Este lote tiene nombres de puesto diferentes y tres particiones en vez de cuatro. Mantén los mismos archivos de programa. El analizador selecciona su tema y catálogo reales, y el descubrimiento encuentra su conjunto completo de particiones reales:
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Debes obtener seis registros físicos, cuatro votos lógicos y amber=2, violet=2, moss=0. Fijar el primer catálogo en el código o detenerse en el número de registros del primer tema no describiría este lote.
Vuelve al primer archivo real:
bigdata-python -B read_snapshot.py output/main-snapshot.json
Todavía contiene el resultado del estudio. El lector abre directamente el archivo solicitado y no aporta nada desde Kafka. Si indica que falta un archivo, compara el nombre con vote_tool.py list y corrige la ruta. Reconstruye el lote correspondiente solo cuando su resultado necesite realmente una reconstrucción.
Rechaza un conflicto y conserva la instantánea válida
En este paso, observarás un error de datos sin sustituir un resultado útil. Continúa en Terminal, en /home/labex/project. El tema conflict de la lista contiene deliberadamente un ID con contenido de puesto válido diferente. Su destino de salida es la instantánea principal existente. No hay un valor ganador aprobado: debe impedirse que esta ejecución publique un resultado como válido.
La huella del paso anterior describe el archivo principal válido. Ejecuta el mismo programa con el lote conflictivo. Se espera que el siguiente comando falle:
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
Debes ver un estado de salida distinto de cero y Conflicting event_id identificando el ID real. La regla se ejecuta antes de que el escritor preparado cree/sustituya el resultado. El escritor con .pending/sustitución es aquí una técnica para publicar resultados completos, no una garantía de recuperación ante caídas ni de durabilidad.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
El archivo principal válido debe seguir coincidiendo, con los recuentos originales del estudio. El conflicto fallido se registra por separado de las ejecuciones correctas de main/check. No edites el origen conflictivo ni elijas el evento que llegó primero. Un recorrido inverso o secuencial de las particiones también debe rechazar el mismo conflicto.
Ante un nombre de lote mal escrito, el punto de entrada muestra los nombres disponibles antes de iniciar la relectura. Para recuperarte, consulta vote_tool.py list y ejecuta el nombre correcto. Para un error local del programa, corrige el archivo y vuelve a ejecutar el lote válido afectado. Tras cualquier edición del código fuente, vuelve a ejecutar ambos lotes válidos para que sus resultados pertenezcan a los archivos actuales; después repite la autocomprobación del conflicto.
Mantén pausados los grupos de las aplicaciones de visualización originales: no está permitido unirse, confirmar, restablecer ni eliminar. Restablecer uno y devolverlo al mismo offset final también queda fuera de una relectura independiente. Las comprobaciones funcionales combinan recepción nativa por el árbol del programa y salidas reales completas; no demuestran causalidad del cálculo ni resistencia frente a un administrador que falsifique el entorno.
Resumen
Has capturado límites reales por partición y leído los logs finitos completos pese a sondeos inactivos transitorios, particiones vacías y posiciones omitidas. El mismo programa nativo reconstruyó dos catálogos sin acumular totales anteriores, y un ID conflictivo provocó un fallo conservando la instantánea principal útil y el progreso original del consumidor.
