Windows do processador de fluxo
Nesta página
As janelas do Atlas Stream Processing são estágios de pipeline de agregação que capturam subconjuntos de tempo limitado de um fluxo de dados, permitindo executar operações que exigem entradas finitos em dados de streaming.
Considere o exemplo de processador de stream descrito aqui. O estágio $match pode operar diretamente no fluxo de dados extraídos por $source, verificando cada documento em relação aos critérios de correspondência à medida que o processador de fluxo o ingere.
Por outro lado, o estágio $group e os vários cálculos estatísticas contidos nele não podem operar em dados ilimitados, pois é impossível determinar valores mínimos, máximos, médios ou medianos sem primeiro limitar o conjunto de valores a serem considerados. Muitos operadores não matemáticos, como $push e $top , também exigem dados limitados.
Um processador de fluxo fornece esses limites com uma janela. Uma janela é aberta e todos os documentos que o processador de fluxo ingere se acumulam no estado dessa janela até que um intervalo de tempo predefinido se esgote e a janela seja fechada. A janela agrupa todos os documentos capturados durante esse intervalo e passa esse conjunto por seu pipeline interno. A partir desse pipeline, os documentos em lote não podem ser distinguidos dos dados em repouso.
Atlas Stream Processing oferece suporte para Tbling Windows e Hopping Windows.
Windows girando
As Windows em queda são Windows definidas inteiramente pelos intervalos de tempo que capturam. Esses intervalos de tempo não se sobrepõem.
Exemplo
Você define uma janela em cascata com um intervalo de 3 segundos. Quando você inicia o processador de stream:
Uma janela abre por 3 segundos.
A primeira janela captura todos os documentos que o fluxo gera dentro desses 3 segundos.
Após 3 segundos decorridos, a janela fecha e aplica sua lógica de agregação a todos os documentos nessa janela.
Se você configurar
allowedLateness, o Atlas Stream Processing gravará mensagens que chegam tarde na dead letter queue (DLQ) após o fechamento da janela.Uma nova janela abre assim que a primeira é fechada e captura documentos do fluxo pelos próximos 3 segundos.
As Windows em cascata garantem a captura abrangente de fluxos de dados sem o processamento repetido de documento individuais.
Windows no salto
Windows de salto são Windows definidas pelo intervalo de tempo que capturam e pelo intervalo entre a abertura de cada janela, chamado de salto. Como a duração é dissociada da frequência, você pode configurar o salto do Windows para se sobrepor ou se separar.
Para definir uma janela de salto com sobreposição, defina um salto menor que o intervalo.
Exemplo
Você define uma janela de salto com um intervalo de 20 segundos e um salto de 5 segundos. Quando você inicia o processador de stream:
Uma janela abre por 20 segundos.
A primeira janela captura todos os documentos que o fluxo gera dentro desses 20 segundos.
5 segundos depois, outra janela abre e captura todos os documentos nos próximos 20 segundos. Como a primeira Windows ainda está aberta, todos os documento gerados pelo fluxo nos próximos 15 segundos são capturados por ambas as Windows.
20 segundos após a abertura da primeira janela, ela fecha e aplica sua lógica de agregação a todos os documentos nessa janela.
5 segundos depois, a segunda janela fecha e aplica sua lógica de agregação a todos os documentos nessa janela, incluindo aqueles que já estavam sujeitos à lógica de agregação na primeira janela.
Se você configurar allowedLateness, o Atlas Stream Processing gravará mensagens que chegam tarde na dead letter queue (DLQ) após o fechamento da janela.
Para definir uma janela de salto com espaçamento, defina um salto maior que o intervalo.
Exemplo
Você define uma janela de salto com um intervalo de 3 segundos e um salto de 5 segundos. Quando você inicia um processador de stream:
Uma janela abre por 3 segundos.
A primeira janela captura todos os documentos nos próximos 3 segundos.
Após 3 segundos decorridos, a janela fecha e aplica sua lógica de agregação a todos os documentos nessa janela.
A próxima janela é aberta após mais 2 segundos.
O Atlas Stream Processing não processa nenhum documento gerado pelo stream durante esses 2 segundos.
Tempo Atlas Stream Processing
No processamento de dados de streaming, os documentos estão sujeitos a dois sistemas de tempo:
Hora do evento - A hora em que o fluxo de origem gera um documento ou o sistema de mensagens (por exemplo, Apache Kafka) recebe o documento. Isso é verificado pelo carimbo de data/hora do documento.
Tempo de processamento - O tempo em que o processador de fluxo consome um documento. Isso é verificado pelo relógio do sistema que hospeda o processador de fluxo.
A latência da rede, o processamento upstream e outros fatores podem não apenas causar discrepâncias entre esses tempos em um determinado documento, mas também podem fazer com que os documentos cheguem em um processador de fluxo fora da ordem do tempo de evento. Em ambos os casos, as janelas podem perder documentos que você pretende que elas capturem. O Atlas Stream Processing considera esses documentos atrasos e os envia para a fila de letras mortas , se você configurar uma.
O Atlas Stream Processing oferece vários mecanismos que alteram o comportamento da janela para mitigar esses problemas.
Marcas d'agua
Uma marca dágua substitui o tempo de processamento e atualiza somente quando o processador consome um documento com um tempo de evento posterior ao de qualquer documento consumido anteriormente. Todos os processadores de stream aplicam marcas d'agua no Atlas Stream Processing.
Exemplo
Você configura um processador de stream com o Windows de 5minutos. Você inicia o processador em 12:00, de modo que as duas primeiras Windows tenham durações de 12:00-12:05 e 12:05-12:10. A tabela a seguir ilustra qual Windows capturará quais eventos dados atrasos variáveis, com e sem marcas d''d'gua.
Hora do evento | Tempo de processamento | Tempo de janela (sem marcas d'gua) | Tempo de janela (marcas d'gua) |
|---|---|---|---|
12:00 | 12:00 | 12:00-12:05 | 12:00-12:05 |
12:00 | 12:01 | 12:00-12:05 | 12:00-12:05 |
12:01 | 12:03 | 12:00-12:05 | 12:00-12:05 |
12:03 | 12:04 | 12:00-12:05 | 12:00-12:05 |
12:02 | 12:05 | 12:05-12:10 | 12:00-12:05 |
12:01 | 12:06 | 12:05-12:10 | 12:00-12:05 |
12:04 | 12:06 | 12:05-12:10 | 12:00-12:05 |
12:05 | 12:07 | 12:05-12:10 | 12:05-12:10 |
12:06 | 12:07 | 12:05-12:10 | 12:04-12:10 |
12:06 | 12:08 | 12:05-12:10 | 12:05-12:10 |
No cenário em que as marcas d'gua não se aplicam, a janela 12:00-12:05 fecha às 12:05, de acordo com o relógio do sistema da instância do Atlas Stream Processing , e a janela 12:05-12:10 abre imediatamente. Como resultado, embora a origem tenha gerado sete dos documentos durante o intervalo 12:00-12:05 , a janela relevante captura apenas quatro documentos.
No cenário em que as marcas d'gua se aplicam, a janela 12:00-12:05 não fecha às 12:05 porque, entre os documentos que ingere até esse ponto, a hora mais recente do evento — e, portanto, o valor da marca d'gua — é 12:03. A janela 12:00-12:05 não fecha até 12:07 no relógio do sistema, quando o processador de fluxo ingere um documento com uma hora de evento de 12:05, avança a marca dágua para essa hora e abre a janela 12:05-12:10 . Cada janela captura todos os documentos apropriados.
Atrasos permitidos
Se as diferenças entre o tempo do evento e o tempo de processamento variarem o suficiente, os documentos poderão chegar a um processador de fluxo depois que a marca d''d'''d''gua tiver avançado o suficiente para fechar a janela esperada. Para mitigar isso, o Atlas Stream Processing suporta o Lateness permitido, uma configuração que atrasa o fechamento de uma janela por um intervalo definido em relação à marca d'gua.
Enquanto as marcas dágua são propriedades de processadores de fluxo, o Lateness permitido é uma propriedade de uma janela e só afeta quando essa janela é fechada. Se a marca dágua do processador de stream avançar para um ponto que acionaria a abertura de uma nova janela, o Lateness permitido manterá as janelas anteriores abertas sem impedir isso.
Exemplo
Você configura um processador de fluxo com 5minutos girando Windows. Você inicia o processador em 12:00, de modo que as duas primeiras Windows tenham durações de 12:00-12:05 e 12:05-12:10. Você definiu um atraso permitido de 2 minutos.
A tabela abaixo reflete a ordem em que o processador de fluxo ingere os documentos descritos.
Hora do evento | Marca D'Água | Tempo de atraso permitido | Janela de tempo |
|---|---|---|---|
12:00 | 12:00 | 11:58 | 12:00-12:05 |
12:01 | 12:01 | 11:59 | 12:00-12:05 |
12:03 | 12:03 | 12:01 | 12:00-12:05 |
12:02 | 12:03 | 12:01 | 12:00-12:05 |
12:04 | 12:04 | 12:02 | 12:00-12:05 |
12:01 | 12:04 | 12:02 | 12:00-12:05 |
12:05 | 12:05 | 12:03 | 12:00-12:15, 12:05-12:10 |
12:06 | 12:06 | 12:04 | 12:00-12:05, 12:05-12:10 |
12:04 | 12:06 | 12:04 | 12:00-12:05, 12:05-12:10 |
12:07 | 12:07 | 12:05 | 12:05-12:10 |
Quando a marca d'gua avança para 12:05, a janela 12:05-12:10 abre. No entanto, como o intervalo de Atraso permitido é 2 minutos, de dentro da janela 12:00-12:05 , ele é efetivamente apenas 12:03, portanto permanece aberto. Somente quando a marca d'gua avança para 12:07 o tempo ajustado atinge 12:05. Neste ponto, a janela 12:00-12:05 fecha.
Tempo-limite de inatividade
Desacoplar o comportamento de janelamento do tempo de processamento por padrão melhora a correção do processamento de fluxo na maioria dos casos. No entanto, uma fonte de dados de streaming pode ter períodos de inatividade prolongado. Nesse cenário, uma janela pode capturar eventos anteriores ao período de ocioso e não conseguir retornar resultados processados enquanto espera que a marca d'''d'''d'ia avance o suficiente para fechar.
O Atlas Stream Processing permite que os usuários configurem um tempo limite de inatividade para o Windows para mitigar esses cenários usando o tempo de processamento. Um tempo limite de ociosidade é um intervalo de tempo que começa quando o tempo de processamento passa do final do intervalo de uma janela aberta e a fonte do processador de fluxo está ociosa. Se a fonte permanecer ociosa por um intervalo igual ao tempo limite de ociosidade, a janela será fechada e a marca d''d'''''ausência independente de qualquer ingestão de documento.
Exemplo
Você configura uma janela de queda com um intervalo de 3minutos e um tempo limite de inatividade 1minutos. A tabela a seguir ilustra os efeitos do tempo limite de inatividade durante e após o intervalo de uma janela.
Tempo de processamento | Hora ou Status do Evento | Marca D'Água | Janela de tempo |
|---|---|---|---|
12:00 | 12:00 | 12:00 | 12:00-12:03 |
12:01 | Fonte ociosa | 12:00 | 12:00-12:03 |
12:02 | Fonte ociosa | 12:00 | 12:00-12:03 |
12:03 | Fonte ociosa | 12:00 | 12:00-12:03 |
12:04 | 12:02 | 12:02 | 12:00-12:03 |
12:05 | 12:05 | 12:05 | 12:03-12:06 |
12:06 | Fonte ociosa | 12:05 | 12:03-12:06 |
12:07 | Fonte ociosa | 12:00 | 12:06-12:09 |
12:08 | Fonte ociosa | 12:00 | 12:06-12:09 |
12:09 | 12:09 | 12:09 | 12:09-12:12 |
Durante o intervalo 12:00-12:03 , a origem fica ociosa por três minutos, mas o processador de fluxo não fecha a janela porque o tempo de processamento não passou do fim do intervalo da janela, e a origem não permanece inativa após o término do intervalo da janela. Quando a marca dágua avança para 12:05, a janela fecha normalmente e a janela 12:03-12:06 abre.
Quando a fonte fica inativa em 12:06, ela permanece inativa em 12:07, acionando o tempo limite de inatividade e avançando a marca d' d'gua para 12:06.