Kafka Stream and KTable One-to-Many Relationship Join

user_4196883 0
Viewed 3355

I have a kafka stream - say for blogs and a kafka table - say for comments related to those blogs. Key from kafka stream can map to multiple values in Kafka table i.e. one blog can have multiple comments. I want to do a join of these two and create a new object with an array of comment ids. But when I do the join, the stream contains only the last comment id. Is there any documentation or example code which can point me right direction how to achieve this? Basically, is there any documentation elaborating how to do one to many relationship join using Kafka stream and Kafka table? 

KStream<Integer, EnrichedBlog> joinedBlogComments = blogsStream.join(commentsTbl,
              (blogId, blog) -> blog.getBlogId(),
              (blog, comment) -> new EnrichedBlog(blog, comment));

So instead of comment - I need to have an array of comment ids.

apache-kafka-streams apache-kafka
3 Answers

If you are using avro with schema registry, you should write your own aggregator because kafka stream fails to serialize ArrayList.

    val kTable = aStream
        .groupByKey()
        .aggregate(
                {
                    YourAggregator() // initialize aggregator
                },
                { _, value, agg ->
                    agg.add(value) // add value to a list in YourAggregator
                    agg
                }
        )

And then join the kTable with your other stream (bStream).

    bStream
        .join(
                kTable,
                { b, a ->
                    // do your value join from a to b
                    b
                }
        )

Sorry my snippets are written in Kotlin.

user_55489260

As pointed out in the correct answer of Michal above, a KTable keyed by blogId cannot be used to keep track of the blogs in this case since only the latest blog value is retained in such table.

As a suggested optimization to the solution mentioned in his answer, note that keeping an ever growing List in the .aggregate() can potentially become costly in both data size and time if there are a lot of comments per blog. This is because under the hood, each iteration of that aggregation leads to ever-growing instances of a List, which is ok in java or scala because of data re-use, but which are each serialized separately to the underlying state-store. Schematically, assuming that some key has say 10 comments, then this expression is called 10 times:

(key, value, agg) -> new KeyValue<>(key, agg.add(value))

each time producing a list of size 1, then 2, then... ... then 10, each serialized independently to the under the hood state store, meaning that 1+2+3+...+10=55 values will be serialized in total (well, maybe there's some optimization s.t. some of those serializations are skipped, I don't know, although the space and time complexity is the same I think).

An alternative, though more complex, approach is to use range scans in state stores, which makes the data structure look a bit like (partition_key, sort_key) in key-value stores like DynamoDB, in which we store each comment with a key like (blogId, commentId). In that case you would still keyBy() the comments stream by blogId, then .transform(...) it to pass it to the processor API, where you can apply the range scan idea, each time adding (i.e. serializing) one single supplementary comment to the state store instead of a new instance of the whole list.

The one-to-many relationship becomes very visible when we picture a lot instances of (blogId, commentId) keys, all having the same blogId and a different commentId, all stored in the same state store instance in the same physical node, and this whole thing happening in parallel for a lot of blogId in a lot of nodes.

I put more details about that pattern on my blog: One-to-many Kafka Streams Ktable join, and I put a full working example in github

user_33183350
Related