Início / coding / Perguntas / 79562196
Accepted
simonalexander2005
Asked: 2025-04-08 21:35:55 +0800 CST

KGroupedStream reduz o uso da chave na lógica do redutor

  • 772

Estou usando o Kafka Streams para agrupar e reduzir um tópico do Kafka.

Quero gerar uma saída para uma chave, se a chave e o valor forem iguais para todos os valores de uma determinada chave; caso contrário, não gere nada para essa chave.

Para fazer isso, no entanto, preciso acessar o key, e os .reducemétodos só dão acesso aos valores agregados/atuais.

É possível obter a chave ao chamar .reduce? Ou existe outra maneira de fazer isso?

apache-kafka

1 respostas

  1. Best Answer

    Você pode usar o aggregatemétodo da KGroupedStreaminterface:

    <VR> KTable<K,VR> aggregate(Initializer<VR> initializer, Aggregator<? super K,? super V,VR> aggregator)

    Agrega os valores dos registros neste fluxo pela chave agrupada. Registros com nullchave ou valor são ignorados. Agregar é uma generalização, combining via reduce(...)pois, por exemplo, permite que o resultado tenha um tipo diferente dos valores de entrada.

    Se você quiser gerar um resultado para uma chave somente se o valor corresponder, crie uma Ktable, algo como:

    KTable<String, Boolean> resultTable = inputStream
    .groupByKey()
    .aggregate(
        () -> true,
        (key, value, agg) -> agg && key.equals(value),
        Materialized.with(Serdes.String(), Serdes.Boolean())
    );

    Você pode então filtrar os valores na tabela, algo como:

    KStream<String, Boolean> finalResult = resultTable
    .toStream()
    .filter((key, allEqual) -> allEqual);
    • 1

relate perguntas