La venta que llegó tarde
El tiempo de evento es el momento en que algo ocurrió de verdad. No siempre coincide con el momento en que el sistema lo recibe. Contar por ventanas de tiempo de evento da la cifra correcta aunque los eventos lleguen desordenados.
Una cadena cuenta cada minuto cuántos productos venden sus cajas. Una caja pierde la red unos segundos y luego envía una venta de las 10:00:58 después de una de las 10:01:02. ¿A qué minuto pertenece? Hoy cuentas de las dos maneras y ves la diferencia.
Al final de esta lección
Nota
Puedo contar eventos en ventanas fijas de tiempo de evento y decir cuáles llegan tarde.
Palabras clave
| Palabra | Qué significa |
|---|---|
| Tiempo de evento | el momento en que la cosa ocurrió de verdad |
| Tiempo de procesamiento | el momento en que el sistema recibió el evento |
| Ventana fija | un trozo de tiempo que nunca se solapa con el siguiente |
| Evento tardío | un evento que llega después de uno más reciente |
| Retraso permitido | el tiempo que una ventana espera a los eventos tardíos |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Todo evento tiene dos relojes
El tiempo de evento es el momento en que algo ocurrió, por ejemplo el instante de una venta. El tiempo de procesamiento es el momento en que tu sistema recibió el evento.
Casi siempre están cerca. Una caída de red, un móvil sin cobertura o un servidor saturado pueden separarlos. Kafka guarda los eventos en el orden de llegada: sus desplazamientos siguen el tiempo de procesamiento, no el de evento.
Ejemplo: cinco ventas
| Desplazamiento | Producto | Tiempo de evento (segundos) | Llegada |
|---|---|---|---|
| 0 | pan | 5 | primera |
| 1 | leche | 62 | segunda |
| 2 | cafe | 58 | tercera, después de 62 |
| 3 | arroz | 125 | cuarta |
| 4 | huevos | 70 | quinta, después de 125 |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Crear un topic de una partición
curl -s -X POST -H "Content-Type: application/json" -d '{"name":"cajas","partitions":1,"replicationFactor":1,"configs":{}}' http://kafka-ui:8080/api/clusters/lab/topics -o /dev/null -w "%{http_code}\n"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Enviar las cinco ventas por orden de llegada
for e in "5 pan" "62 leche" "58 cafe" "125 arroz" "70 huevos"; do curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"caja\",\"value\":\"$e\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages -w "%{http_code}\n"; doneSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Ver los desplazamientos
curl -s http://kafka-ui:8080/api/clusters/lab/topics/cajas | grep -o '"partition":[0-9]*\|"offsetMax":[0-9]*'Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Leer el topic desde el principio
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"offset":[0-9]*\|"value":"[^"]*"'Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Contar los minutos de reloj del broker
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"timestamp":"[^"]*"' | cut -c25-29 | sort -u | wc -lSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Una ventana corta el tiempo de evento
Para contar por minuto, cortas el tiempo en ventanas de 60 segundos. Una ventana fija empieza donde acaba la anterior: cada evento está en una sola.
El inicio de la ventana de un evento es su tiempo menos el resto de dividir entre el tamaño. El tiempo 62 da 62 menos 2: la ventana empieza en 60.
Ejemplo: ventanas de 60 segundos
| Ventana | Empieza | Termina antes de | Productos |
|---|---|---|---|
| Primera | 0 | 60 | pan, cafe |
| Segunda | 60 | 120 | leche, huevos |
| Tercera | 120 | 180 | arroz |
Las celdas que rellenan los estudiantes quedan en blanco en esta vista previa.
Contar los eventos de cada ventana
# Cada evento es [tiempo del evento en segundos, producto], en el orden en que llegó.
events = [[5, "pan"], [62, "leche"], [58, "cafe"], [125, "arroz"], [70, "huevos"]]
size = 60
counts = {}
for event in events:
start = event[0] // size * size
if start in counts:
counts[start] = counts[start] + 1
else:
counts[start] = 1
print(counts)
# Cambia size a 30 y vuelve a ejecutarlo.
Contar las ventanas desde el propio topic
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Un error habitual
Advertencia
Mucha gente cree que el orden del topic es el orden en que pasaron las cosas. Es el orden de llegada. Usa el tiempo escrito en cada evento para elegir su ventana, y los desplazamientos solo para decir qué has leído.
Un evento tardío llega después de otro
El café se vendió a los 58 segundos, pero llegó después de la leche de los 62. Es un evento tardío: uno más reciente llegó primero.
Un sistema real no puede esperar para siempre. Cierra una ventana cuando el tiempo más reciente que ha visto pasa de su final. Un pequeño retraso permitido mantiene la ventana abierta un poco más.
Encontrar los eventos tardíos
# Un evento llega tarde si antes de él llegó un evento con un tiempo mayor.
events = [[5, "pan"], [62, "leche"], [58, "cafe"], [125, "arroz"], [70, "huevos"]]
newest = 0
for event in events:
if event[0] < newest:
print(event[1], "llega tarde: su tiempo", event[0], "es anterior a", newest)
else:
newest = event[0]
# Añade [30, "te"] al final de events y vuelve a ejecutarlo.
Cerrar una ventana tras un retraso permitido
# Una ventana se cierra cuando el tiempo más reciente alcanza su final más el retraso permitido.
events = [[5, "pan"], [62, "leche"], [58, "cafe"], [125, "arroz"], [70, "huevos"]]
size = 60
lateness = 5
newest = 0
counts = {}
dropped = []
for event in events:
start = event[0] // size * size
if newest >= start + size + lateness:
dropped.append(event[1])
elif start in counts:
counts[start] = counts[start] + 1
else:
counts[start] = 1
if event[0] > newest:
newest = event[0]
print(counts, dropped)
# Cambia lateness a 0, luego a 10, y vuelve a ejecutarlo.
Enviar otra venta, muy tardía
curl -s -X POST -H "Content-Type: application/json" -d "{\"partition\":0,\"key\":\"caja\",\"value\":\"30 te\",\"headers\":{},\"keySerde\":\"String\",\"valueSerde\":\"String\"}" http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages -w "%{http_code}\n"Se muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Volver a contar las ventanas
curl -s -N "http://kafka-ui:8080/api/clusters/lab/topics/cajas/messages/v2?seekType=BEGINNING&limit=10" | grep -o '"value":"[^"]*"' | cut -d'"' -f4 | while read t item; do echo $((t / 60)); done | sort -n | uniq -cSe muestra aquí y no se ejecuta. Crea una cuenta gratuita para pulsar Ejecutar y escribirlo en el terminal de tu propio laboratorio.
Para ir más lejos: elegir un retraso
Vuelve a ejecutar el código del retraso con 0, 5 y 10. Anota cuántos eventos descarta cada uno.
Después responde en una frase: un informe de cajas para el director al final del día y una alarma que debe saltar en segundos. ¿Cuál puede esperar a los eventos tardíos y por qué?