如何在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());
}
您通过将所有要分组的属性/字段放入密钥中来指定组合密钥。
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