Apache Kafka 1.0.0流媒体API多个多级集团

Question

如何在kafka streams api中使用多个约束。与Java 8流中的API示例相同

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());
}

Answer 1

您通过将所有要分组的属性/字段放入密钥中来指定组合密钥。

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

我只是以为国家和城市都是String。您使用交互式查询与

查询商店

store.get("UK-London");

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