卡夫卡流分组依据和串联

Question

我有一个接收记录的 Kafka 流，我想根据特定字段连接消息。流中的消息如下所示：Key: 2099Payload{  email: [e-mail]  eventCode: 2099}预期输出：key: 2099Payload{    emails: tom@example, [e-mail], [e-mail]}我可以让溪流运行良好，我只是不确定lamda应该包含什么。这就是我迄今为止所做的。我不确定我是否应该使用映射，聚合或减少或组合这些操作。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 );

Answer 1

它可能是这样的东西

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