merge records in a kafka stream

user_1042429 0
Viewed 1324

Is it possible to merge records in kafka and publish the output to different stream ?

For example , there is a stream of events produced to a kafka topic like below

{txnId:1,startTime:0900},{txnId:1,endTime:0905},{txnId:2,endTime:0912},{txnId:3,endTime:0930},{txnId:2,startTime:0912},{txnId:3,startTime:0925}......

I want to merge these events by txnId and create the merged output like below

{txnId:1,startTime:0900,endTime:0905},{txnId:2,startTime:0910,endTime:0912},{txnId:3,startTime:0925,endTime:0930}

Please note that order is not maintained in the incoming events.So if endTime is received for a txn Id before start time event , then we need to wait till the start time event is received for that txnId before initiating the merge

I went through the word count example that comes along with Kafka Streams example but its not clear how to wait for events and then merge while doing the transformation.

Any thoughts is highly appreciated.

confluent-platform apache-kafka-streams kafka-consumer-api apache-kafka
1 Answers

You could try solving this by splitting the start and end events into 2 separate streams with txnId as the key and then joining both the streams.

KStream<String, String> eventSource = new StreamBuilder().stream("INPUT-TOPIC");

KStream<String, JsonNode>[] splitEvents = 
          eventSource.map((key, eventString) -> {
                           JsonNode event = new ObjectMapper().readTree(eventString);
                           String txnId = event.path("txnId").asText();
                           return KeyValue.pair(txnId, event);
                        })
                     .branch((key, event) -> event.findValue("startTime") != null,
                             (key, event) -> event.findValue("endTime") != null); 


KStream<String, JsonNode> startEvents = splitEvents[0];
KStream<String, JsonNode> endEvents = splitEvents[1];

A join between 2 streams as shown will produce a join result when there is an event in either side of the join. So the order of both events won't matter (you will have to ensure that you set an appropriate window period for the join).

Serde<JsonNode> jsonSerde = Serdes.serdeFrom(new JsonSerializer(), new JsonDeserializer());

KStream<String, String> completeEvents = startEvents.join(endEvents, 
               (startEvent, endEvent) -> {
                    // Add logic to merge startEvent and endEvent as seen fit
                       ObjectNode completeEvent = JsonNodeFactory.instance.objectNode();
                       completeEvent.put("startTime",  startEvent.path("startTime).asText());
                       completeEvent.put("endTime",  endEvent.path("endTime").asText());
                       return completeEvent.toString();
                },
               JoinWindows.of(Duration.ofMinutes(15)),
               Joined.with(
                    Serdes.String(),   // key
                    jsonSerde,         // left object
                    jsonSerde          // right object
               )
          );
user_66307620
Related