Kafka Streams DSL: Aggregating a KStream to a GlobalKTable

user_1461196 0
Viewed 525

I have an input stream:

KStream<String, X> inputStream = ...

That I would like to operate on (filter then aggregate) in such a way that I output into a GlobalKTable<String, Y> which I can then read back in using:

KeyValueIterator<String, Y> = streams.store("y-global-store", QueryableStoreTypes.keyValueStore()).all()

Can the Streams DSL support this? It seems it would be possible if the output table was a KTable, but given the small amount of data I will have in this store I'd like to use a GlobalKTable

Here is my processor which converts a KStream<String, X> to a KTable<String, Y>

KTable<String, Y> outputTable = inputStream
    .filter(...)
    .groupByKey(Grouped.with(Serdes.String(), ySerde))
    .aggregate(
        initializeWithNull(),
        aggregateXToAY(),
        Materialized.`as`<String, Y, KeyValueStore<Bytes, ByteArray>>("y-global-store")
            .withKeySerde(stringSerde)
            .withValueSerde(tagRecordSerde)
    )

However this doesn't create a GlobalKTable, what am I missing?

apache-kafka-streams
1 Answers

Stream DSL does not support building GlobalKTable from a KStream. It seems that the only way you could create a GlobalKTable is to using StreamsBuilder. globalTable("input_topic_for_globalktable")

I think the reason the DSL does not support create GlobalKtable this way is that each application instance contain the whole GlobalKTable state, so it's disabled logging by default (it does not log changelog to the changelog topic), so it uses an input topic directly for restore state process (fault tolerance), this topic must have log compaction enabled.

One solution is you have to prepare data for this input topic before as output of outputTable KTable:

outputTable.toStream().to("input_topic_for_globalktable");

or directly using the changelog topic of outputTable KTable (I think this solution is better since you do not need additional disk space for the new topic), it has the name :

<your-application_id>-y-global-store-changelog
user_84568620
Related