CODEWITHSUNDEEP
javaapache-kafkaapache-kafka-streams
Niraj Sanghani
Niraj Sanghani

Reputation: 1493

Apache Kafka 1.0.0 Streams API Multiple Multilevel groupby

How do i use .groupby with Multiple constrainst in Kafka Streams API. Same as in Java 8 Streams API example below 

public void twoLevelGrouping(List<Person> persons) {
     final Map<String, Map<String, List<Person>>> personsByCountryAndCity = persons.stream().collect(
         groupingBy(Person::getCountry,
            groupingBy(Person::getCity)
        )
    );
    System.out.println("Persons living in London: " + personsByCountryAndCity.get("UK").get("London").size());
}

Upvotes: 7

Views: 1326

Answers (1)

Matthias J. Sax
Matthias J. Sax

Reputation: 62350

You specify a combined key by putting all attributes/fields you want to group-by into the key.

KTable table = stream.selectKey((k, v,) -> k::getCountry + "-" + k::getCity)
                     .groupByKey()
                     .aggregate(...); // or maybe .reduce()

I just assumed that country and city are both String. You use Interactive Queries to query the store with

store.get("UK-London");

https://docs.confluent.io/current/streams/developer-guide/interactive-queries.html

Upvotes: 4

Related Questions