$throttle Stage de agregación (Stream Processing)

Definición

$throttle

La etapa $throttle limita la velocidad a la que un procesador de flujo transfiere datos a las etapas posteriores. Utilice $throttle para proteger los sistemas posteriores de picos de tráfico y para mantenerse dentro de los límites de velocidad que imponen dichos sistemas.

Los picos de tráfico pueden producirse cuando un procesador de flujo toma una instantánea inicial de un conjunto de datos grande, se recupera tras una interrupción o reprocesa datos históricos. Sin un límite de velocidad, estos picos pueden sobrecargar el sistema de destino, obligarlo a aumentar su escala o superar la cuota de una API de terceros.

Sintaxis

Un $throttle etapa tiene la siguiente forma prototipo:

{
  $throttle: {
    bytesPerSec: <integer>,
    messagesPerSec: <integer>
  }
}

La etapa $throttle procesa un documento con los siguientes campos:

Debe especificar al menos uno de bytesPerSec o messagesPerSec. Si no especifica ninguno, Atlas Stream Processing devuelve el siguiente error:

$throttle requires at least one of 'bytesPerSec' or 'messagesPerSec'

Comportamiento

Atlas Stream Processing realiza un seguimiento de un límite de velocidad independiente para cada campo que usted configure. Cada límite recupera capacidad continuamente a la velocidad que usted configure, y Atlas Stream Processing realiza un seguimiento de cada límite de forma independiente.

Los mensajes solo pueden pasar de la etapa $throttle cuando todos los límites configurados tienen capacidad disponible. Si se configuran ambos campos, ambos límites deben tener capacidad. Cuando se agota más de un límite, la etapa espera el tiempo que requiera el límite más agotado. Todos los límites recuperan capacidad durante esa espera, por lo que el límite con menor déficit se recupera antes de que finalice la espera.

La etapa no espera un segundo completo antes de transmitir más datos. Procesa los siguientes mensajes tan pronto como cada límite configurado tenga capacidad suficiente para ellos.

Si un documento individual supera el límite bytesPerSec, la etapa no lo retiene hasta que se acumule suficiente capacidad. En su lugar, la etapa pasa el documento a la siguiente etapa y espera un segundo antes de pasar más datos.

Puedes usar más de una etapa $throttle en una misma canalización. Cada etapa limita únicamente los datos que la atraviesan, lo que te permite aplicar límites diferentes a distintas partes de la canalización.

Puedes usar $throttle solo en la canalización principal. No puedes anidar $throttle dentro de otra etapa.

Monitoring

Atlas Stream Processing informa las siguientes estadísticas para los procesadores de flujo que utilizan $throttle:

Atlas Stream Processing informa estadísticas tanto a nivel de procesador como para cada etapa. Cuando una canalización contiene más de una etapa $throttle, los valores a nivel de procesador son la suma de los valores por etapa. Las estadísticas por etapa son stats.operatorStats.throttle.throttledTimeMs y stats.operatorStats.throttle.throttleEvents.

Para obtener más información sobre las estadísticas del procesador de flujos, consulte la sección "Supervisión y métricas del procesamiento de flujos de Atlas".

Ejemplos

La siguiente canalización limita los datos que un procesador de flujo escribe en un tema de Kafka a 5 MiB y 50 mensajes por segundo:

{
  $throttle: {
    bytesPerSec: 5242880,
    messagesPerSec: 50
  }
},
{
  $emit: {
    connectionName: "ordersTopic"
  }
}

Consideremos un lote de 20 documentos que suman un total de 2 MiB. El tráfico reciente ha utilizado parte de la capacidad de ambos límites:

La etapa no puede continuar hasta que ambos límites tengan suficiente capacidad, por lo que espera 200 ms, el mayor de los dos tiempos de espera. Dado que ambos límites recuperan capacidad durante esa espera, el límite messagesPerSec, que solo necesitaba 100 ms, ya se ha recuperado cuando finaliza la espera.