O livro Kafka Streams em ação. Aplicativos e microsserviços em tempo real »

Oi, habrozhiteli! Este livro é adequado para qualquer desenvolvedor que queira entender o processamento de streaming. Compreender a programação distribuída ajudará você a entender melhor o Kafka e o Kafka Streams. Seria bom conhecer o próprio framework Kafka, mas isso não é necessário: vou lhe contar tudo o que você precisa. Graças a este livro, desenvolvedores experientes do Kafka, como novatos, aprenderão como criar aplicativos interessantes de streaming usando a biblioteca Kafka Streams. Desenvolvedores Java intermediários e de alto nível familiarizados com conceitos como serialização aprenderão como aplicar suas habilidades para criar aplicativos Kafka Streams. O código-fonte do livro é escrito em Java 8 e usa essencialmente a sintaxe das expressões lambda do Java 8, portanto, a capacidade de trabalhar com funções lambda (mesmo em outra linguagem de programação) é útil.

Trecho. 5.3 Agregação e operações de janela

Nesta seção, passamos às partes mais promissoras do Kafka Streams. Até agora, cobrimos os seguintes aspectos do Kafka Streams:

• criando uma topologia de processamento;
• uso do estado em aplicativos de streaming;
• fazer conexões de fluxo de dados;
• diferenças entre fluxos de eventos (KStream) e fluxos de atualização (KTable).

Nos exemplos a seguir, colocaremos todos esses elementos juntos. Além disso, você será apresentado às operações da janela - outro ótimo recurso dos aplicativos de streaming. Nosso primeiro exemplo será agregação simples.

5.3.1 Agregação de vendas de ações por setor

A agregação e o agrupamento são ferramentas vitais para trabalhar com dados de streaming. Examinar registros individuais com base na admissão geralmente não é suficiente. Para extrair informações adicionais dos dados, seu agrupamento e combinação são necessários.

Neste exemplo, você deve experimentar o processo de um trader intradiário que precisa acompanhar o volume de vendas de ações de empresas de vários setores. Em particular, você está interessado nas cinco empresas com as maiores vendas de ações em cada setor.

Para essa agregação, você precisará de várias das etapas a seguir para converter os dados no formato desejado (em termos gerais).

1. Crie uma fonte baseada em tópicos que publique informações brutas sobre negociação de ações. Teremos que mapear um objeto do tipo StockTransaction para um objeto do tipo ShareVolume. O fato é que o objeto StockTransaction contém metadados de vendas e precisamos apenas de dados sobre o número de compartilhamentos vendidos.
2. Agrupe os dados do ShareVolume por símbolos de ações. Após o agrupamento por símbolos, você pode recolher esses dados em subtotais de vendas de ações. Vale ressaltar que o método KStream.groupBy retorna uma instância do tipo KGroupedStream. E você pode obter uma instância do KTable chamando o método KGroupedStream.reduce posteriormente.

O que é a interface KGroupedStream

Os métodos KStream.groupBy e KStream.groupByKey retornam uma instância de KGroupedStream. KGroupedStream é uma representação intermediária do fluxo de eventos após o agrupamento por chave. Não se destina a trabalhar diretamente com ele. Em vez disso, o KGroupedStream é usado para operações de agregação, cujo resultado é sempre o KTable. E como o resultado das operações de agregação é o KTable e elas usam armazenamento de estado, é possível que nem todas as atualizações sejam enviadas mais adiante no pipeline.

O método KTable.groupBy retorna um KGroupedTable semelhante - uma representação intermediária do fluxo de atualizações reagrupadas por chave.

Vamos fazer uma pequena pausa e olhar para a fig. 5.9, que mostra o que alcançamos. Essa topologia já deve ser familiar para você.


Agora, vamos dar uma olhada no código para esta topologia (ela pode ser encontrada no arquivo src / main / java / bbejeck / chapter_5 / AggregationsAndReducingExample.java) (Listagem 5.2).


O código fornecido difere na brevidade e um grande volume de ações executadas em várias linhas. No primeiro parâmetro do método builder.stream, é possível observar algo novo para você: o valor do tipo enumerado AutoOffsetReset.EARLIEST (também existe LATEST), configurado usando o método Consumed.withOffsetResetPolicy. Usando esse tipo enumerado, você pode especificar uma estratégia para redefinir compensações para cada um dos KStream ou KTable; ela tem prioridade sobre o parâmetro para redefinir compensações da configuração.

GroupByKey e GroupBy

A interface do KStream possui dois métodos para agrupar registros: GroupByKey e GroupBy. Ambos retornam uma KGroupedTable, para que você possa ter uma pergunta legítima: qual é a diferença entre eles e quando usar qual?

O método GroupByKey é usado quando as chaves no KStream já estão vazias. E o mais importante, o sinalizador "requer re-particionamento" nunca foi definido.

O método GroupBy pressupõe que você alterou as chaves de agrupamento, portanto, o sinalizador de particionamento é definido como verdadeiro. A realização de conexões, agregações etc. após o método GroupBy levará ao particionamento automático.
Resumo: você deve usar GroupByKey em vez de GroupBy sempre que possível.

O que os métodos mapValues ​​e groupBy fazem é compreensível, portanto, dê uma olhada no método sum () (ele pode ser encontrado no arquivo src / main / java / bbejeck / model / ShareVolume.java) (Listagem 5.3).


O método ShareVolume.sum retorna o subtotal do volume de vendas de ações e o resultado de toda a cadeia de cálculo é um objeto KTable <String, ShareVolume>. Agora você entende o papel que o KTable desempenha. Quando os objetos ShareVolume chegam, a atualização atual mais recente é salva na KTable correspondente. É importante não esquecer que todas as atualizações são refletidas no shareVolumeKTable anterior, mas nem todas são enviadas mais.

Além disso, com a ajuda desta KTable, realizamos agregação (pelo número de ações vendidas) para obter as cinco empresas com as maiores vendas de ações em cada setor. Nossas ações nesse caso serão semelhantes às ações durante a primeira agregação.

1. Execute outra operação groupBy para agrupar objetos individuais do ShareVolume por setor.
2. Continue resumindo os objetos ShareVolume. Desta vez, o objeto de agregação é uma fila prioritária de tamanho fixo. Apenas cinco empresas com o maior número de ações vendidas são mantidas em uma fila de tamanho fixo.
3. Exiba as linhas do parágrafo anterior em um valor de sequência e retorne as cinco mais vendidas pelo número de ações por setor.
4. Escreva os resultados em forma de sequência no tópico.

Na fig. 5.10 mostra um gráfico da topologia da movimentação de dados. Como você pode ver, a segunda rodada de processamento é bastante simples.


Agora, tendo entendido claramente a estrutura desta segunda rodada de processamento, você pode consultar o código-fonte (você o encontrará no arquivo src / main / java / bbejeck / chapter_5 / AggregationsAndReducingExample.java) (Listagem 5.4).

Há uma variável FixedQueue neste inicializador. Este é um objeto personalizado - um adaptador para java.util.TreeSet, usado para rastrear N resultados mais altos em ordem decrescente do número de compartilhamentos vendidos.


Você já encontrou chamadas para groupBy e mapValues, por isso não as interromperemos (chamamos o método KTable.toStream, pois o método KTable.print foi descontinuado). Mas você ainda não viu a versão KTable do método agregate (), portanto, passaremos algum tempo discutindo isso.

Como você se lembra, o KTable se distingue pelo fato de os registros com as mesmas chaves serem considerados atualizações. O KTable substitui o registro antigo pelo novo. A agregação ocorre da mesma maneira: os últimos registros com uma chave são agregados. Quando um registro chega, ele é adicionado a uma instância da classe FixedSizePriorityQueue usando um somador (o segundo parâmetro na chamada de método agregado), mas se já existir outro registro com a mesma chave, o registro antigo será excluído usando o subtractor (o terceiro parâmetro na chamada de método agregado).

Isso tudo significa que nosso agregador, FixedSizePriorityQueue, não agrega todos os valores com uma chave, mas armazena a soma móvel das quantidades N dos tipos de ações mais vendidos. Cada entrada contém o número total de ações vendidas até o momento. O KTable fornece informações sobre quais ações das empresas estão sendo mais vendidas no momento; a agregação contínua de cada atualização não é necessária.

Aprendemos a fazer duas coisas importantes:

• agrupe valores no KTable por uma chave comum a eles;
• Execute operações úteis, como convolução e agregação, nesses valores agrupados.

A capacidade de executar essas operações é importante para entender o significado dos dados que se deslocam pelo aplicativo Kafka Streams e descobrir quais informações eles carregam.

Também reunimos alguns dos principais conceitos discutidos anteriormente neste livro. No Capítulo 4, falamos sobre a importância de um estado local à prova de falhas para um aplicativo de streaming. O primeiro exemplo deste capítulo mostrou por que o estado local é tão importante - possibilita rastrear as informações que você já viu. O acesso local evita atrasos na rede, tornando o aplicativo mais produtivo e resistente a erros.

Ao executar qualquer operação de convolução ou agregação, você deve especificar o nome do armazenamento de estado. As operações de convolução e agregação retornam uma instância do KTable, e o KTable usa um armazenamento de estado para substituir resultados antigos por novos. Como você viu, nem todas as atualizações são enviadas mais adiante, e isso é importante, pois as operações de agregação são projetadas para obter as informações finais. Se o estado local não for aplicado, o KTable enviará ainda mais todos os resultados de agregação e convolução.

A seguir, examinamos a execução de operações como agregação, dentro de um período específico de tempo - as chamadas operações de janelas.

5.3.2 Operações de janela

Na seção anterior, introduzimos a convolução e agregação “rolantes”. O aplicativo executou uma convolução contínua das vendas de ações com a agregação subsequente das cinco ações mais vendidas.

Às vezes, essa agregação e convolução contínuas de resultados são necessárias. E às vezes você precisa executar operações apenas em um determinado período de tempo. Por exemplo, calcule quantas transações de bolsa foram feitas com ações de uma empresa específica nos últimos 10 minutos. Ou quantos usuários clicaram em um novo anúncio em banner nos últimos 15 minutos. Um aplicativo pode executar essas operações várias vezes, mas com resultados relacionados apenas a intervalos de tempo especificados (janelas de tempo).

Contando transações de câmbio por comprador

No exemplo a seguir, participaremos do rastreamento de transações de câmbio para vários traders - grandes organizações ou financiadores inteligentes de uma mão.

Há dois motivos possíveis para esse rastreamento. Uma delas é a necessidade de saber quais líderes de mercado estão comprando / vendendo. Se esses grandes players e investidores sofisticados vêem oportunidades para si mesmos, faz sentido seguir sua estratégia. A segunda razão é o desejo de observar possíveis sinais de transações ilegais usando informações privilegiadas. Para fazer isso, você precisará analisar a correlação de grandes picos nas vendas com importantes press releases.

Esse rastreamento consiste em etapas como:

• criando um fluxo para leitura do tópico de transações de estoque;
• agrupamento de registros recebidos por ID do cliente e símbolo do estoque. Uma chamada para o método groupBy retorna uma instância da classe KGroupedStream;
• KGroupedStream.windowedBy retorna um fluxo de dados delimitado por uma janela temporária, o que permite a agregação de janelas. Dependendo do tipo de janela, TimeWindowedKStream ou SessionWindowedKStream é retornado;
• Contando transações para uma operação de agregação. O fluxo de dados da janela determina se um registro específico é levado em consideração nesse cálculo;
• gravando resultados em um tópico ou enviando-os para o console durante o desenvolvimento.

A topologia deste aplicativo é simples, mas sua imagem visual não faz mal. Dê uma olhada na foto. 5.11

Além disso, consideraremos a funcionalidade das operações da janela e o código correspondente.

Tipos de janela

Existem três tipos de janelas no Kafka Streams:

• sessão
• Cair (tombar);
• deslizando / "pulando" (deslizando / pulando).

Qual escolher depende dos requisitos de negócios. As janelas "tombar" e "pular" são limitadas no tempo, enquanto as restrições da sessão estão associadas às ações do usuário - a duração da (s) sessão (s) é determinada apenas pela forma como o usuário se comporta ativamente. O principal é não esquecer que todos os tipos de janelas são baseados em registros de data / hora dos registros e não na hora do sistema.

Em seguida, implementamos nossa topologia com cada um dos tipos de janela. O código completo será fornecido apenas no primeiro exemplo, nada será alterado para outros tipos de janelas, exceto para o tipo de operação da janela.

Janelas de sessão

As janelas de sessão são muito diferentes de todos os outros tipos de janelas. Eles são limitados não tanto pelo tempo como pela atividade do usuário (ou pela atividade da entidade que você deseja rastrear). As janelas da sessão são delimitadas por períodos de inatividade.

A Figura 5.12 ilustra o conceito de janelas de sessão. Uma sessão menor será mesclada com a sessão à esquerda. E a sessão à direita será separada, pois segue um longo período de inatividade. As janelas da sessão são baseadas em ações do usuário, mas aplicam registros de data / hora dos registros para determinar a qual sessão o registro pertence.

Usando o Windows da sessão para rastrear transações do Exchange

Usaremos janelas de sessão para capturar informações sobre transações de câmbio. A implementação das janelas da sessão é mostrada na Listagem 5.5 (que pode ser encontrada em src / main / java / bbejeck / chapter_5 / CountingWindowingAndKTableJoinExample.java).


Você já conheceu a maioria das operações desta topologia, portanto não há necessidade de considerá-las aqui novamente. Mas há vários novos elementos que discutiremos agora.

Para qualquer operação groupBy, geralmente é executado algum tipo de operação de agregação (agregação, convolução ou contagem). Você pode executar a agregação cumulativa com um total cumulativo ou a agregação de janelas, na qual os registros são levados em consideração em uma determinada janela de tempo.

O código na Listagem 5.5 conta o número de transações nas janelas da sessão. Na fig. 5.13 essas ações são analisadas passo a passo.

Ao chamar windowedBy (SessionWindows.with (twentySeconds) .until (fifteenMinutes)), criamos uma janela de sessão com um intervalo ocioso de 20 segundos e um intervalo de retenção de 15 minutos. Um intervalo de inatividade de 20 segundos significa que o aplicativo incluirá qualquer registro que chegue dentro de 20 segundos a partir do final ou do início da sessão atual na sessão atual (ativa).


A seguir, indicamos qual operação de agregação executar na janela da sessão - neste caso, conte. Se o registro recebido estiver fora do intervalo de inatividade (em ambos os lados do registro de data / hora), o aplicativo criará uma nova sessão. O intervalo de salvamento significa manter a sessão por um certo tempo e permitir dados atrasados ​​que vão além do período de inatividade da sessão, mas ainda podem ser anexados. Além disso, o início e o fim de uma nova sessão resultante da mesclagem correspondem ao carimbo de data / hora mais antigo e mais recente.

Vejamos algumas entradas do método count para ver como as sessões funcionam (Tabela 5.1).


Após o recebimento dos registros, procuramos sessões já existentes com a mesma tecla, o horário final é menor que o registro de data / hora atual - o intervalo de inatividade e o horário de início é maior que o registro de data / hora atual + intervalo de inatividade. Com isso em mente, quatro registros da tabela. 5.1 mesclar em uma única sessão da seguinte maneira.

1. O registro 1 vem primeiro, portanto, o horário de início é igual ao horário de término e é 00:00:00.

2. Em seguida, vem o registro 2, e procuramos sessões que terminem antes das 23:59:55 e iniciem até 00:00:35. Localize o registro 1 e combine as sessões 1 e 2. Pegue a hora de início da sessão 1 (anterior) e a hora de término da sessão 2 (posterior), para que nossa nova sessão comece às 00:00:00 e termine às 00:00:15.

3. O registro 3 chega, procuramos sessões entre 00:00:30 e 00:01:10 e não encontramos nenhuma. Adicione uma segunda sessão para a chave 123-345-654, FFBE, iniciando e terminando às 00:00:50.

4. O registro 4 chega e procuramos sessões entre 23:59:45 e 00:00:25. Desta vez, existem duas sessões - 1 e 2. Todas as três sessões são combinadas em uma, com um horário de início de 00:00:00 e um horário de término de 00:00:15.

Pelo que é dito nesta seção, vale lembrar as seguintes nuances importantes:

• As sessões não são janelas de tamanho fixo. A duração de uma sessão é determinada pela atividade dentro de um determinado período de tempo;
• Os carimbos de data / hora nos dados determinam se um evento cai em uma sessão existente ou em um período de inatividade.

Além disso, discutiremos o seguinte tipo de janela - janela "cambalhota".

Janelas em queda

Janelas "caídas" capturam eventos que caem dentro de um determinado período de tempo. Imagine que você precise capturar todas as transações de câmbio de uma empresa a cada 20 segundos, para coletar todos os eventos desse período. No final do intervalo de 20 segundos, a janela "tomba" e muda para um novo intervalo de observação de 20 segundos. A Figura 5.14 ilustra essa situação.


Como você pode ver, todos os eventos recebidos nos últimos 20 segundos estão incluídos na janela. No final desse período, uma nova janela é criada.

A Listagem 5.6 mostra o código que demonstra o uso de janelas alternadas para capturar transações de câmbio a cada 20 segundos (você pode encontrá-lo em src / main / java / bbejeck / chapter_5 / CountingWindowingAndKtableJoinExample.java).


Graças a essa pequena alteração na chamada para o método TimeWindows.of, você pode usar a janela de queda. Neste exemplo, não há chamada para o método till (), como resultado do qual o intervalo de salvamento padrão de 24 horas será usado.

Finalmente, é hora de passar para a última das opções da janela - pulando janelas.

Janelas deslizantes ("pulando")

As janelas deslizantes / “saltitantes” são semelhantes a “tombar”, mas com uma pequena diferença. As janelas deslizantes não esperam o final do intervalo de tempo antes de criar uma nova janela para manipular eventos recentes. Eles iniciam novos cálculos após um intervalo de espera menor que a duração da janela.

Para ilustrar as diferenças entre as janelas "cambalhota" e "saltando", voltemos ao exemplo com o cálculo das transações de câmbio. Nosso objetivo, como antes, é contar o número de transações, mas não queremos esperar o tempo todo antes de atualizar o contador. Em vez disso, atualizaremos o contador em intervalos mais curtos. Por exemplo, continuaremos contando o número de transações a cada 20 segundos, mas atualizando o contador a cada 5 segundos, conforme mostrado na Fig. 5.15 Ao mesmo tempo, temos três janelas de resultados com dados sobrepostos.


A Listagem 5.7 mostra o código para especificar janelas deslizantes (ele pode ser encontrado em src / main / java / bbejeck / chapter_5 / CountingWindowingAndKtableJoinExample.java).


«» «» advanceBy(). 15 .

, . , , :

• , ;
• «» ;
• «» , .

, KTable KStream .

5.3.3. KStream KTable

4 KStream. KTable KStream. . KStream — , KTable — , KTable.

. , .

1. KTable KStream , , .
2. KTable, . KTable .
3. .

, .

KTable KStream

KTable KStream .

1. KTable.toStream().
2. KStream.map , Windowed TransactionSummary.

( src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) ( 5.8).


KStream.map, KStream .

, KTable .

KTable

, KTable ( src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) ( 5.9).


, Serde , Serde. EARLIEST .

— .

. , ( src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) ( 5.10).


leftJoin . 4, JoinWindow , KStream-KTable KTable . : KTable, . : KTable KStream .

KStream.

5.3.4. GlobalKTable

, . 4 KStream, — KStream KTable. . , Kafka Streams . , , ( 4, « » 4.2.4).

— , ; . , , .

, , , . Kafka Streams GlobalKTable.

GlobalKTable , . , , . GlobalKTable . .

KStream GlobalKTable

5.3.2 . :

{customerId='074-09-3705', stockTicker='GUTM'}, 17 {customerId='037-34-5184', stockTicker='CORK'}, 16 

Embora esses resultados fossem consistentes com a meta, seria mais conveniente se o nome do cliente e o nome completo da empresa também fossem exibidos. Para adicionar o nome de um cliente e o nome de uma empresa, você pode realizar conexões normais, mas precisará fazer dois mapeamentos principais e o particionamento novamente. Com o GlobalKTable, você pode evitar o custo de tais operações.

Para fazer isso, usaremos o objeto countStream da Listagem 5.11 (o código correspondente pode ser encontrado no arquivo src / main / java / bbejeck / chapter_5 / GlobalKTableExample.java), conectando-o a dois objetos GlobalKTable.


Já discutimos isso antes, então não vou repetir. Mas observo que o código na função toStream () .Mapa é abstraído no objeto de função para facilitar a legibilidade, em vez da expressão lambda incorporada.

A próxima etapa é declarar duas instâncias do GlobalKTable (o código mostrado pode ser encontrado em src / main / java / bbejeck / chapter_5 / GlobalKTableExample.java) (Listagem 5.12).



Observe que os nomes dos tópicos são descritos usando tipos enumerados.

Agora que preparamos todos os componentes, resta escrever o código da conexão (que pode ser encontrado no arquivo src / main / java / bbejeck / chapter_5 / GlobalKTableExample.java) (Listagem 5.13).


Embora existam dois compostos nesse código, eles são organizados em uma cadeia, pois nenhum de seus resultados é usado separadamente. Os resultados são exibidos no final de toda a operação.

Ao iniciar a operação de conexão acima, você obterá os seguintes resultados:

 {customer='Barney, Smith' company="Exxon", transactions= 17} 

A essência não mudou, mas esses resultados parecem mais claros.

Contando o capítulo 4, você já viu vários tipos de conexões em ação. Eles estão listados na tabela. 5.2 Esta tabela reflete a conectividade relevante para a versão 1.0.0 do Kafka Streams; algo mudará em versões futuras.


Concluindo, lembrarei o principal: você pode conectar fluxos de eventos (KStream) e atualizar fluxos (KTable) usando o estado local. Além disso, se o tamanho dos dados de referência não for muito grande, você poderá usar o objeto GlobalKTable. O GlobalKTable replica todas as seções em cada um dos nós do aplicativo Kafka Streams, garantindo a disponibilidade de todos os dados, independentemente da seção à qual a chave corresponde.

A seguir, veremos a possibilidade do Kafka Streams, graças à qual é possível observar alterações de estado sem consumir dados do tópico Kafka.

5.3.5 Status da solicitação

Já realizamos várias operações envolvendo o estado e sempre produzimos os resultados no console (para fins de desenvolvimento) ou escrevemos no tópico (para operação industrial). Ao gravar resultados em um tópico, você deve usar o consumidor Kafka para visualizá-los.

A leitura de dados desses tópicos pode ser considerada um tipo de visão materializada. Para nossas tarefas, podemos usar a definição de uma visão materializada da Wikipedia: “... um objeto de banco de dados físico contendo os resultados de uma consulta. Por exemplo, pode ser uma cópia local dos dados excluídos, um subconjunto de linhas e / ou colunas de uma tabela ou resultados de junção ou uma tabela dinâmica obtida usando agregação ”(https://en.wikipedia.org/wiki/Materialized_view).

O Kafka Streams também permite executar consultas interativas em lojas de estado, o que permite ler diretamente essas visualizações materializadas. É importante observar que a solicitação para o armazenamento de estado é da natureza de uma operação somente leitura. Graças a isso, você não pode ter medo de tornar acidentalmente o estado um aplicativo inconsistente durante o processamento de dados.

A capacidade de consultar diretamente os armazenamentos de estado é importante. Isso significa que você pode criar aplicativos - painéis sem precisar primeiro receber dados de um consumidor Kafka. Aumenta a eficiência do aplicativo, devido ao fato de não ser necessário registrar dados novamente:

• devido à localidade dos dados, eles podem ser acessados ​​rapidamente;
• A duplicação de dados é excluída, pois eles não são gravados no armazenamento externo.

A principal coisa que eu gostaria que você se lembrasse: você pode executar diretamente solicitações de estado do aplicativo. Você não pode superestimar as oportunidades que isso oferece. Em vez de consumir dados do Kafka e armazenar registros no banco de dados do aplicativo, você pode consultar os armazenamentos de estado com o mesmo resultado. Solicitações diretas para lojas de estado significam menos código (sem consumidor) e menos software (sem necessidade de uma tabela de banco de dados para armazenar os resultados).

Abordamos uma quantidade considerável de informações neste capítulo; portanto, interromperemos temporariamente nossa discussão de consultas interativas para lojas de estado. Mas não se preocupe: no capítulo 9, criaremos um aplicativo simples - um painel de informações com consultas interativas. Para demonstrar consultas interativas e as possibilidades de adicioná-las aos aplicativos Kafka Streams, ele utilizará alguns dos exemplos deste e dos capítulos anteriores.

Sumário

• Os objetos KStream representam fluxos de eventos comparáveis ​​às inserções do banco de dados. Os objetos KTable representam fluxos de atualização, são mais semelhantes às atualizações no banco de dados. O tamanho do objeto KTable não aumenta; registros antigos são substituídos por novos.
• Os objetos KTable são necessários para operações de agregação.
• Usando as operações da janela, você pode dividir os dados agregados em cestas de horas.
• Graças aos objetos GlobalKTable, você pode acessar dados de referência em qualquer lugar do aplicativo, independentemente do corte.
• Conexões entre os objetos KStream, KTable e GlobalKTable são possíveis.

Até o momento, focamos na criação de aplicativos Kafka Streams usando o DSL de alto nível do KStream. Embora uma abordagem de alto nível permita a criação de programas puros e concisos, seu uso é um compromisso definitivo. Trabalhar com o DSL KStream significa aumentar a concisão do código, reduzindo o grau de controle. No próximo capítulo, examinaremos a API de baixo nível dos nós manipuladores e tentaremos outras compensações. Os programas ficarão mais longos do que eram até agora, mas teremos a oportunidade de criar quase todos os nós de processamento que precisarmos.

→ Mais detalhes sobre o livro podem ser encontrados no site do editor

→ Para Khabrozhiteley 25% de desconto no cupom - Kafka Streams

→ Após o pagamento da versão impressa do livro, um livro eletrônico é enviado por e-mail.