Hi Patrik,
Thanks for explaining your use case to us. While we can still discuss how
KStream should interpret null-values in aggregations, one workaround atm:
if you deduplication logic can be written as a transformValues operation,
you can do the following:
builder.table("source-topic").transformValues(...
Materialized.as("store-name"))
Note that in a recent PR that we are merging, the source KTable from
builder.table() would not be materialized if users do not specify a
materialized store name, only the value-transformed KTable will be
materialized:
https://github.com/apache/kafka/pull/5779
Would that work for you?
Guozhang
On Mon, Oct 29, 2018 at 2:08 AM Patrik Kleindl <pklei...@gmail.com> wrote:
> 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
> > > >>>>
> > > >>>
> > > >>
> > > >
> > >
> > >
> >
>
--
-- Guozhang
