Introducción
El productor de consola ha demostrado el modelo de eventos y claves. Una aplicación necesita una distinción más: llamar a una API de envío puede colocar un registro en una cola local del cliente antes de que el bróker lo confirme. Un programa que termine en ese momento podría anunciar que ha tenido éxito sin haber entregado sus registros.
Utilizará el cliente Python confluent-kafka instalado con el bróker local de Apache Kafka preparado. El paquete es una biblioteca cliente; no necesita una cuenta de Confluent Cloud. Serializará diccionarios como bytes de JSON en UTF-8, enviará claves de pedido e inspeccionará las funciones de confirmación de entrega y el estado final de la cola. Una dirección local inaccesible permitirá probar un fallo con tiempo limitado.
Este laboratorio independiente prepara un tema de tres particiones y dos archivos de entrada. Las actividades posteriores del curso presuponen conocimientos de funciones, bucles, diccionarios y JSON en Python, así como de ejecución de scripts; la ruta de Python puede ayudarle con esos requisitos previos. La instalación de Kafka y la configuración del servicio ya están preparadas. Utilice la Terminal.
Convierta un diccionario en una clave y un valor para transmitir
En este paso inspeccionará la conversión antes de contactar con Kafka. Comience con la entrada de seis eventos:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Cada línea es un objeto JSON completo. json.loads convierte ese texto en un diccionario de Python. json.dumps convierte el diccionario en texto JSON, y encode("utf-8") convierte el texto en bytes adecuados para la API del cliente. La clave se codifica por separado a partir de order_id; no es event_id.
Guarde y ejecute esta pequeña vista previa. El documento incrustado con delimitador entre comillas conserva literalmente el código Python:
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Localice el diccionario, el texto JSON y las representaciones en bytes. La representación de bytes de Python empieza por b; esa notación no forma parte del valor transmitido. La primera clave debe codificar order-4001. JSON conserva los campos con nombre del evento, mientras que UTF-8 proporciona la codificación de caracteres.
Todavía no se ha enviado ningún evento. El siguiente paso suministra estos bytes a un productor y comprueba cuándo los confirma el bróker.
Publique y confirme cada entrega real
En este paso ejecutará un productor completo con gestión de argumentos y una función de confirmación de entrega suministradas. Producer configura el cliente; bootstrap.servers proporciona la dirección inicial. acks=all solicita la confirmación de las réplicas actualmente sincronizadas. Este entorno de un solo bróker tiene una réplica, por lo que ese ajuste no demuestra tolerancia a fallos entre varios brókeres. El ajuste de enrutamiento murmur2_random suministrado mantiene la compatibilidad del enrutamiento con clave con los ejemplos anteriores de consola Java mientras la cantidad de particiones permanece fija.
Observe la distinción fundamental del programa: produce encola un registro localmente. poll atiende las funciones de confirmación, y flush espera al trabajo encolado a la vez que atiende esas funciones.
Una función de retorno (callback) es una función que proporciona a la biblioteca cliente para que la invoque cuando haya un resultado de entrega disponible. Pasar callback=delivery proporciona la propia función; no la llama inmediatamente. poll(0) atiende las funciones de retorno disponibles sin esperar, mientras que flush también espera a los envíos pendientes. La función registra un tema, partición y desplazamiento confirmados por el bróker o un fallo. Después de flush, el programa comprueba los registros restantes, los fallos y la cantidad confirmada antes de anunciar que ha tenido éxito. Los fallos de entrega se recopilan y provocan un resultado de proceso distinto de cero; no se limitan a mostrarse para luego ignorarse.
La sección argparse suministrada asigna opciones de línea de comandos a atributos de args: por ejemplo, --input se convierte en args.input, y --delivery-timeout-ms en args.delivery_timeout_ms. Path(args.input).open(...) abre el archivo local seleccionado; el bucle lee un evento JSON por cada línea no vacía. No necesita diseñar un analizador de argumentos para utilizar este cliente.
Guarde el programa completo. La estructura del analizador y de la función de retorno ya se proporciona, mientras que la serialización y la llamada a produce utilizan los bytes que acaba de inspeccionar:
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Ejecute el productor y guarde ambos flujos de salida. > dirige la salida estándar al archivo de registro; 2>&1 envía el error estándar al mismo destino, de modo que también se conserven los fallos de las funciones de retorno y los diagnósticos de conexión.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Deben aparecer seis líneas DELIVERED de las funciones de retorno y Delivered records: 6. La partición y el desplazamiento reales de cada confirmación pueden variar. Que produce termine correctamente no demostraría por sí solo estas confirmaciones.
Inspeccione el bróker de forma independiente con el consumidor de consola que ya conoce:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Localice de py1 a py6 con claves que coincidan con sus valores order_id. La lectura independiente vincula el informe de entrega de la aplicación con los registros reales del bróker; no se exige ningún número de partición fijo ni un orden global de presentación.
Cambie un campo de un evento nuevo y reutilice el productor
En este paso reutilizará la misma aplicación para un evento nuevo. El archivo next-event preparado tiene un event_id nuevo, py7, por lo que aún no se ha publicado. Inspeccione su campo channel:
cat /home/labex/project/data/next-event.jsonl
Cambie únicamente el channel de ese evento no publicado de online a store en otro archivo de trabajo. El evento sigue correspondiendo a order-4001 y continúa sus eventos anteriores con sequence 3:
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Debe observar una nueva confirmación de entrega correcta y siete eventos retenidos en la lectura independiente. py7 debe tener channel store, mientras que de py1 a py6 conservan su contenido original. Los eventos relacionados de order-4001 siguen su secuencia dentro de la misma partición asignada; el orden de presentación entre particiones puede variar.
La implementación del productor no ha cambiado. La ruta de entrada suministrada le ha permitido serializar y entregar los datos nuevos en lugar de incorporar una respuesta fija del ejercicio en el código.
Observe un fallo con tiempo limitado en una dirección inaccesible
En este paso comprobará que encolar no se confunde con entregar. El mismo cliente puede crear una cola local aunque en localhost:1 no haya ningún bróker de Kafka. La opción de tiempo de espera de entrega suministrada limita la finalización del mensaje a 3000 milisegundos, y flush añade un pequeño margen acotado. El bróker real sigue en localhost:9092.
Envíe el archivo de un evento a esa dirección deliberadamente incorrecta. Se espera que el siguiente comando falle.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Capture el estado inmediatamente, antes de que otro comando lo sustituya. $? es el estado de salida del comando anterior: cero indica éxito y un valor distinto de cero, fallo. printf lo guarda como una línea de texto:
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Debe observar un estado distinto de cero, diagnósticos de conexión, una llamada de retorno DELIVERY_FAILED y Delivery incomplete en lugar de un recuento final satisfactorio. El programa ha comprobado los fallos de las funciones de retorno incluso cuando la cola ha terminado de vaciarse tras fallar los mensajes.
La dirección fallida no debe haber añadido un segundo py7 al tema real. Los siete eventos entregados correctamente siguen disponibles. Corregir una dirección y reintentar un evento de negocio es una decisión distinta de suponer que un registro encolado ya ha sido confirmado; el fallo deliberado de este laboratorio no ha enviado nada al bróker real. A continuación, decodificará el JSON recibido, lo guardará y confirmará el progreso de lectura en un consumidor Python.
Resumen
Ha serializado diccionarios y claves de pedido como bytes UTF-8, publicado mediante un cliente Python real y relacionado las confirmaciones correctas con posiciones reales del bróker. Un evento nuevo modificado ha reutilizado el mismo programa, mientras que una dirección inaccesible ha producido un fallo con tiempo limitado y estado distinto de cero en lugar de un falso éxito. A continuación, un consumidor Python hará explícitas la decodificación, la salida a archivos y la escritura antes de confirmar el progreso.
