kafka流分组和级联

rjee0c15 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(532)

我有一个接收记录的kafka流，我想根据特定字段连接消息。
流中的消息如下所示：

Key: 2099
Payload{
  email: tom@emample.com
  eventCode: 2099
}

预期产量：

key: 2099
Payload{
    emails: tom@example, bill@acme.com, jane@example.com
}

我可以让这条小溪流淌得很好，我只是不知道兰达河应该包含什么。
这就是我到目前为止所做的。我不确定是否应该使用map、aggregate或reduce或这些操作的组合。

final StreamsBuilder builder = new StreamsBuilder();
KStream<String, Payload> inputStream = builder.stream(INPUT_TOPIC);

inputStream
        .groupByKey()
        .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(300000)))

                                  // Not sure what to do here …..

}).to (OUTPUT_TOPIC );

Java apache-kafka apache-kafka-streams

来源：https://stackoverflow.com/questions/55497032/kafka-streams-group-by-and-concatenation

1条答案

按热度按时间

xtfmy6hx1#

可能是这样的

inputStream.groupByKey().windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(300000)))
.aggregate(PayloadAggr::new, new Aggregator<String, Payload, PayloadAggr>() {
        @Override
        public PayloadAggr apply(String key, Payload newValue, PayloadAggr result) {
            result.setKey(key);
            if(result.getEmails()==null){
                result.setEmails(newValue.getEmail());
            }else{
                result.setEmails(result.getEmails() + "," + newValue.getEmail());
            }
            return result;
        }
    }, .../* You serdes and store */}).toStream().to(OUTPUT_TOPIC);

赞(0）回复(0）举报 2021-06-07

我来回答

kafka流分组和级联

1条答案

相关问题

热门标签

最新问答