$throttle Estágio de agregação (processamento de fluxo)

Definição

$throttle

O estágio $throttle limita a taxa na qual um processador de stream passa dados para os estágios seguintes. Use $throttle para proteger os sistemas downstream de explosões de tráfego e para permanecer dentro dos limites de taxa que esses sistemas impõem.

As intermitências podem ocorrer quando um processador de stream tira um snapshot inicial de um grande conjunto de dados, se recupera após uma interrupção ou reprocessa dados históricos. Sem um limite de taxa, essas intermitências podem sobrecarregar o sistema de destino, forçá-lo a escalar ou exceder uma cota de API de terceiros.

Sintaxe

Um estágio $throttle tem o seguinte formato de protótipo:

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

O estágio $throttle recebe um documento com os seguintes campos:

Você deve especificar pelo menos um entre bytesPerSec ou messagesPerSec. Se você não especificar nenhum dos dois, o Atlas Stream Processing retornará o seguinte erro:

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

Comportamento

O Atlas Stream Processing rastreia um limite de taxa separado para cada campo que você define. Cada limite recupera a capacidade continuamente na taxa que você configura, e o Atlas Stream Processing rastreia cada limite independentemente dos outros.

As mensagens podem passar por um estágio $throttle somente quando cada limite configurado tiver capacidade disponível para elas. Se você definir ambos os campos, ambos os limites deverão ter capacidade. Quando mais de um limite está esgotado, o estágio espera o tempo que o limite mais esgotado exige. Todos os limites recuperam a capacidade durante essa espera, portanto, um limite com um fracasso menor se recupera antes que a espera termine.

O estágio não espera um segundo inteiro antes de passar mais dados. Ele processa as próximas mensagens assim que cada limite configurado tiver capacidade suficiente para elas.

Se um único documento for maior que o limite bytesPerSec, o estágio não reterá o documento até que capacidade suficiente acumule. Em vez disso, o estágio passa o documento para o próximo estágio e espera um segundo antes de passar mais dados.

Você pode usar mais de um estágio $throttle em um único pipeline. Cada estágio limita apenas os dados que fluem por ele, o que permite aplicar limites diferentes a diferentes partes do pipeline.

Você pode usar $throttle somente no pipeline principal. Você não pode aninhar $throttle dentro de outro estágio.

Monitoramento

O Atlas Stream Processing informa as seguintes estatísticas para processadores de stream que usam $throttle:

O Atlas Stream Processing relata as estatísticas no nível do processador e para cada estágio. Quando um pipeline contém mais de um estágio $throttle, os valores no nível do processador são a soma dos valores por estágio. As estatísticas por estágio são stats.operatorStats.throttle.throttledTimeMs e stats.operatorStats.throttle.throttleEvents.

Para saber mais sobre estatísticas de processador de fluxo, consulte Monitoramento e Métricas de Atlas Stream Processing .

Exemplos

O pipeline a seguir limita os dados que um processador de fluxo grava em um tópico do Kafka a 5 MiB e 50 mensagens por segundo:

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

Considere um lote de 20 documentos totalizando 2 MiB. O tráfego recente usou parte da capacidade de ambos os limites:

O estágio não pode prosseguir até que ambos os limites tenham capacidade suficiente, então ele espera 200 ms, o mais longo dos dois tempos de espera. Como ambos os limites recuperam a capacidade durante a espera, o limite messagesPerSec, que precisava de apenas 100 ms, já se recuperou quando a espera termina.