Ideal way to perform lookup on a stream of Kafka topic

user_7027284 0
Viewed 894

I have the following use-case: There is a stream of records on a Kafka topic. I have another set of unique IDs. I need to, for each record in the stream, check if the stream's ID is present in the set of unique IDs I have. Basically, this should serve as a filter for my Kafka Streams app. i.e., only to write records of Kafka topic that match the set of Unique IDs I have to another topic.

Our current application is based on Kafka Streams. I looked at KStreams and KTables. Looks like they're good for enrichments. Now, I don't need any enrichments to the data. As for using state stores, I'm not sure how good they are as a scalable solution.

I would like to do something like this:

kStream.filter((k, v) -> {
    valueToCheckInKTable = v.get(FIELD_NAME);
    if (kTable.containsKey(valueToCheckInKTable)) return record
    else ignore 
});

The lookup data can be pretty huge. Can someone suggest the best way to do this?

apache-kafka-streams apache-kafka
1 Answers

You can read the reference IDs into a table via builder.table("id-topic") with the ID as primary key (note that the value must be non-null -- otherwise it would be interpreted as a delete -- if there is no actual value, just put any non-null dummy value of each record when you write the IDs into the id-topic). To load the full table on startup, you might want to provide a custom timestamp extractor that always returns 0 via Consumed parameter on the table() operator (record are processed in timestamp order and returning 0 ensure that the record from the id-topic are processed first to load the table).

To do the filtering you do a stream-table join:

KStream stream = builder.stream(...);
// the key in the stream must be ID, if this is not the case, you can use `selectKey()` to set a new ke

KStream filteredStream = stream.join(table,...);

As you don't want to do any enrichment, the provided Joiner function can just return the left stream side value unmodified (and can ignored the right hand side table value).

user_49530790
Related