Hi John and Matthias
thanks for the questions, maybe explaining our use case helps a bit:
We are receiving CDC records (row-level insert/update/delete) in one topic
per table. The key is derived from the DB records, the value is null in
case of deletes. Those would be the immutable facts I guess.
These topics are first streamed through a deduplication Transformer to drop
changes on irrelevant fields.
The results are translated to KTables and joined to each other to represent
the same result as the SQLs on the database, but faster. At this stage the
delete/null records matter because if a record gets deleted then we want it
to drop out of the join too. -> Our reduce-approach produced unexpected
results here.
We took the deduplication step separately because in some cases we only
need the the KStream for processing.
If you see a simpler/cleaner approach here I'm open to suggestions, of
course.
Regarding the overhead:
1) Named topics create management/maintenance overhead because they have to
be created/treated separately (auto-create is not an option) and be
considered in future changes, topology changes/resets and so on. The
internal topic removes most of those issues.
2) One of our developers came up with the question if the traffic to/from
the broker was actually the same in both scenarios, we expect that the same
is written to the broker for the named topic as well as the reduce-case,
but if the KTable is maintained inside a streams topology, does it have to
read back everything it sends to the broker or can it keep the table
internally? I hope it is understandable what I mean, otherwise I can try
the explain it more clearly.
best regards
Patrik
On Sat, 27 Oct 2018 at 23:50, John Roesler <j...@confluent.io> wrote:
> Hi again Patrik,
>
> Actually, this is a good question... Can you share some context about why
> you need to convert a stream to a table (including nulls as retractions)?
>
> Thanks,
> -John
>
> On Fri, Oct 26, 2018 at 5:36 PM Matthias J. Sax <matth...@confluent.io>
> wrote:
>
> > I don't know your overall application setup. However, a KStream
> > semantically models immutable facts and there is not update semantic.
> > Thus, it seems semantically questionable, to allow changing the
> > semantics from facts to updates (the other way is easier IMHO, and thus
> > supported via KTable#toStream()).
> >
> > Does this make sense?
> >
> > Having said this: you _can_ write a KStream into a topic an read it back
> > as KTable. But it's semantically questionable to do so, IMHO. Maybe it
> > makes sense for your specific application, but in general I don't think
> > it does make sense.
> >
> >
> > -Matthias
> >
> > On 10/26/18 9:30 AM, John Roesler wrote:
> > > Hi Patrik,
> > >
> > > Just to drop one observation in... Streaming to a topic and then
> > consuming
> > > it as a table does create overhead, but so does reducing a stream to a
> > > table, and I think it's actually the same in either case.
> > >
> > > They both require a store to collect the table state, and in both
> cases,
> > > the stores need to have a changelog topic. For the "reduce" version,
> it's
> > > an internal changelog topic, and for the "topic-to-table" version, the
> > > store can use the intermediate topic as its changelog.
> > >
> > > This doesn't address your ergonomic concern, but it seemed worth
> pointing
> > > out that (as far as I can tell), there doesn't seem to be a difference
> in
> > > overhead.
> > >
> > > Hope this helps!
> > > -John
> > >
> > > On Fri, Oct 26, 2018 at 3:27 AM Patrik Kleindl <pklei...@gmail.com>
> > wrote:
> > >
> > >> Hello Matthias,
> > >> thank you for the explanation.
> > >> Streaming back to a topic and consuming this as a KTable does respect
> > the
> > >> null values as deletes, correct? But at the price of some overhead.
> > >> Is there any (historical, technical or emotional;-)) reason that no
> > simple
> > >> one-step stream-to-table operation exists?
> > >> Best regards
> > >> Patrik
> > >>
> > >>> Am 26.10.2018 um 00:07 schrieb Matthias J. Sax <
> matth...@confluent.io
> > >:
> > >>>
> > >>> Patrik,
> > >>>
> > >>> `null` values in a KStream don't have delete semantics (it's not a
> > >>> changelog stream). That's why we drop them in the KStream#reduce
> > >>> implemenation.
> > >>>
> > >>> If you want to explicitly remove results for a key from the result
> > >>> KTable, your `Reducer#apply()` implementation must return `null` --
> the
> > >>> result of #apply() has changelog/KTable semantics and `null` is
> > >>> interpreted as delete for this case.
> > >>>
> > >>> If you want to use `null` from your KStream to trigger reduce() to
> > >>> delete, you will need to use a surrogate value for this, ie, do a
> > >>> mapValues() before the groupByKey() call, an replace `null` values
> with
> > >>> the surrogate-delete-marker that you can evaluate in
> `Reducer#apply()`
> > >>> to return `null` for this case.
> > >>>
> > >>> Hope this helps.
> > >>>
> > >>> -Matthias
> > >>>
> > >>>> On 10/25/18 10:36 AM, Patrik Kleindl wrote:
> > >>>> Hello
> > >>>>
> > >>>> Recently we noticed a lot of warning messages in the logs which
> > pointed
> > >> to
> > >>>> this method (we are running 2.0):
> > >>>>
> > >>>> KStreamReduce
> > >>>> public void process(final K key, final V value) {
> > >>>> // If the key or value is null we don't need to proceed
> > >>>> if (key == null || value == null) {
> > >>>> LOG.warn(
> > >>>> "Skipping record due to null key or value.
> key=[{}]
> > >>>> value=[{}] topic=[{}] partition=[{}] offset=[{}]",
> > >>>> key, value, context().topic(),
> > context().partition(),
> > >>>> context().offset()
> > >>>> );
> > >>>> metrics.skippedRecordsSensor().record();
> > >>>> return;
> > >>>> }
> > >>>>
> > >>>> This was triggered for every record from a stream with an existing
> key
> > >> but
> > >>>> a null value which we put through groupBy/reduce to get a KTable.
> > >>>> My assumption was that this was the correct way inside a streams
> > >>>> application to get a KTable but this prevents deletion of records
> from
> > >>>> working.
> > >>>>
> > >>>> Our alternativ is to send the stream back to a named topic and
> build a
> > >> new
> > >>>> table from it, but this is rather cumbersome and requires a separate
> > >> topic
> > >>>> which also can't be cleaned up by the streams reset tool.
> > >>>>
> > >>>> Did I miss anything relevant here?
> > >>>> Would it be possible to create a separate method for KStream to
> > achieve
> > >>>> this directly?
> > >>>>
> > >>>> best regards
> > >>>>
> > >>>> Patrik
> > >>>>
> > >>>
> > >>
> > >
> >
> >
>
