Thanks
Adam
On Tue, Sep 11, 2018 at 10:07 AM, Prajakta Dumbre
<dumbreprajakta...@gmail.com <mailto:dumbreprajakta...@gmail.com>> wrote:
please remove me from this group
On Tue, Sep 11, 2018 at 1:29 PM Jan Filipiak
<jan.filip...@trivago.com <mailto:jan.filip...@trivago.com>>
wrote:
> Hi Adam,
>
> give me some time, will make such a chart. last time i didn't
get along
> well with giphy and ruined all your charts.
> Hopefully i can get it done today
>
> On 08.09.2018 16:00, Adam Bellemare wrote:
> > Hi Jan
> >
> > I have included a diagram of what I attempted on the KIP.
> >
>
https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+Support+non-key+joining+in+KTable#KIP-213Supportnon-keyjoininginKTable-GroupBy+Reduce/Aggregate
<https://cwiki.apache.org/confluence/display/KAFKA/KIP-213+Support+non-key+joining+in+KTable#KIP-213Supportnon-keyjoininginKTable-GroupBy+Reduce/Aggregate>
> >
> > I attempted this back at the start of my own implementation of
this
> > solution, and since I could not get it to work I have since
discarded the
> > code. At this point in time, if you wish to continue pursuing
for your
> > groupBy solution, I ask that you please create a diagram on
the KIP
> > carefully explaining your solution. Please feel free to use
the image I
> > just posted as a starting point. I am having trouble
understanding your
> > explanations but I think that a carefully constructed diagram
will clear
> up
> > any misunderstandings. Alternately, please post a
comprehensive PR with
> > your solution. I can only guess at what you mean, and since I
value my
> own
> > time as much as you value yours, I believe it is your
responsibility to
> > provide an implementation instead of me trying to guess.
> >
> > Adam
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > On Sat, Sep 8, 2018 at 8:00 AM, Jan Filipiak
<jan.filip...@trivago.com <mailto:jan.filip...@trivago.com>>
> > wrote:
> >
> >> Hi James,
> >>
> >> nice to see you beeing interested. kafka streams at this
point supports
> >> all sorts of joins as long as both streams have the same key.
> >> Adam is currently implementing a join where a KTable and a
KTable can
> have
> >> a one to many relation ship (1:n). We exploit that rocksdb is a
> >> datastore that keeps data sorted (At least exposes an API to
access the
> >> stored data in a sorted fashion).
> >>
> >> I think the technical caveats are well understood now and we are
> basically
> >> down to philosophy and API Design ( when Adam sees my newest
message).
> >> I have a lengthy track record of loosing those kinda
arguments within
> the
> >> streams community and I have no clue why. So I literally
can't wait for
> you
> >> to churn through this thread and give you opinion on how we
should
> design
> >> the return type of the oneToManyJoin and how many power we
want to give
> to
> >> the user vs "simplicity" (where simplicity isn't really that
as users
> still
> >> need to understand it I argue)
> >>
> >> waiting for you to join in on the discussion
> >>
> >> Best Jan
> >>
> >>
> >>
> >> On 07.09.2018 15:49, James Kwan wrote:
> >>
> >>> I am new to this group and I found this subject
interesting. Sounds
> like
> >>> you guys want to implement a join table of two streams? Is there
> somewhere
> >>> I can see the original requirement or proposal?
> >>>
> >>> On Sep 7, 2018, at 8:13 AM, Jan Filipiak
<jan.filip...@trivago.com <mailto:jan.filip...@trivago.com>>
> >>>> wrote:
> >>>>
> >>>>
> >>>> On 05.09.2018 22:17, Adam Bellemare wrote:
> >>>>
> >>>>> I'm currently testing using a Windowed Store to store the
highwater
> >>>>> mark.
> >>>>> By all indications this should work fine, with the caveat
being that
> it
> >>>>> can
> >>>>> only resolve out-of-order arrival for up to the size of
the window
> (ie:
> >>>>> 24h, 72h, etc). This would remove the possibility of it being
> unbounded
> >>>>> in
> >>>>> size.
> >>>>>
> >>>>> With regards to Jan's suggestion, I believe this is where
we will
> have
> >>>>> to
> >>>>> remain in disagreement. While I do not disagree with your
statement
> >>>>> about
> >>>>> there likely to be additional joins done in a real-world
workflow, I
> do
> >>>>> not
> >>>>> see how you can conclusively deal with out-of-order arrival of
> >>>>> foreign-key
> >>>>> changes and subsequent joins. I have attempted what I
think you have
> >>>>> proposed (without a high-water, using groupBy and reduce)
and found
> >>>>> that if
> >>>>> the foreign key changes too quickly, or the load on a
stream thread
> is
> >>>>> too
> >>>>> high, the joined messages will arrive out-of-order and be
incorrectly
> >>>>> propagated, such that an intermediate event is represented
as the
> final
> >>>>> event.
> >>>>>
> >>>> Can you shed some light on your groupBy implementation.
There must be
> >>>> some sort of flaw in it.
> >>>> I have a suspicion where it is, I would just like to
confirm. The idea
> >>>> is bullet proof and it must be
> >>>> an implementation mess up. I would like to clarify before
we draw a
> >>>> conclusion.
> >>>>
> >>>> Repartitioning the scattered events back to their original
> >>>>> partitions is the only way I know how to conclusively deal
with
> >>>>> out-of-order events in a given time frame, and to ensure
that the
> data
> >>>>> is
> >>>>> eventually consistent with the input events.
> >>>>>
> >>>>> If you have some code to share that illustrates your
approach, I
> would
> >>>>> be
> >>>>> very grateful as it would remove any misunderstandings
that I may
> have.
> >>>>>
> >>>> ah okay you were looking for my code. I don't have
something easily
> >>>> readable here as its bloated with OO-patterns.
> >>>>
> >>>> its anyhow trivial:
> >>>>
> >>>> @Override
> >>>> public T apply(K aggKey, V value, T aggregate)
> >>>> {
> >>>> Map<U, V> currentStateAsMap = asMap(aggregate); <<
imaginary
> >>>> U toModifyKey = mapper.apply(value);
> >>>> << this is the place where people actually
gonna have
> issues
> >>>> and why you probably couldn't do it. we would need to find
a solution
> here.
> >>>> I didn't realize that yet.
> >>>> << we propagate the field in the joiner, so
that we can
> pick
> >>>> it up in an aggregate. Probably you have not thought of
this in your
> >>>> approach right?
> >>>> << I am very open to find a generic solution
here. In my
> >>>> honest opinion this is broken in KTableImpl.GroupBy that it
looses
> the keys
> >>>> and only maintains the aggregate key.
> >>>> << I abstracted it away back then way before i was
> thinking
> >>>> of oneToMany join. That is why I didn't realize its
significance here.
> >>>> << Opinions?
> >>>>
> >>>> for (V m : current)
> >>>> {
> >>>> currentStateAsMap.put(mapper.apply(m), m);
> >>>> }
> >>>> if (isAdder)
> >>>> {
> >>>> currentStateAsMap.put(toModifyKey, value);
> >>>> }
> >>>> else
> >>>> {
> >>>> currentStateAsMap.remove(toModifyKey);
> >>>> if(currentStateAsMap.isEmpty()){
> >>>> return null;
> >>>> }
> >>>> }
> >>>> retrun asAggregateType(currentStateAsMap)
> >>>> }
> >>>>
> >>>>
> >>>>
> >>>>
> >>>>
> >>>> Thanks,
> >>>>> Adam
> >>>>>
> >>>>>
> >>>>>
> >>>>>
> >>>>>
> >>>>> On Wed, Sep 5, 2018 at 3:35 PM, Jan Filipiak <
> jan.filip...@trivago.com <mailto:jan.filip...@trivago.com>>
> >>>>> wrote:
> >>>>>
> >>>>> Thanks Adam for bringing Matthias to speed!
> >>>>>> about the differences. I think re-keying back should be
optional at
> >>>>>> best.
> >>>>>> I would say we return a KScatteredTable with reshuffle()
returning
> >>>>>> KTable<originalKey,Joined> to make the backwards
repartitioning
> >>>>>> optional.
> >>>>>> I am also in a big favour of doing the out of order
processing using
> >>>>>> group
> >>>>>> by instead high water mark tracking.
> >>>>>> Just because unbounded growth is just scary + It saves us
the header
> >>>>>> stuff.
> >>>>>>
> >>>>>> I think the abstraction of always repartitioning back is
just not so
> >>>>>> strong. Like the work has been done before we partition
back and
> >>>>>> grouping
> >>>>>> by something else afterwards is really common.
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>> On 05.09.2018 13:49, Adam Bellemare wrote:
> >>>>>>
> >>>>>> Hi Matthias
> >>>>>>> Thank you for your feedback, I do appreciate it!
> >>>>>>>
> >>>>>>> While name spacing would be possible, it would require to
> deserialize
> >>>>>>>
> >>>>>>>> user headers what implies a runtime overhead. I would
suggest to
> no
> >>>>>>>> namespace for now to avoid the overhead. If this becomes a
> problem in
> >>>>>>>> the future, we can still add name spacing later on.
> >>>>>>>>
> >>>>>>>> Agreed. I will go with using a reserved string and
document it.
> >>>>>>>
> >>>>>>>
> >>>>>>> My main concern about the design it the type of the
result KTable:
> If
> >>>>>>> I
> >>>>>>> understood the proposal correctly,
> >>>>>>>
> >>>>>>>
> >>>>>>> In your example, you have table1 and table2 swapped.
Here is how it
> >>>>>>> works
> >>>>>>> currently:
> >>>>>>>
> >>>>>>> 1) table1 has the records that contain the foreign key
within their
> >>>>>>> value.
> >>>>>>> table1 input stream: <a,(fk=A,bar=1)>, <b,(fk=A,bar=2)>,
> >>>>>>> <c,(fk=B,bar=3)>
> >>>>>>> table2 input stream: <A,X>, <B,Y>
> >>>>>>>
> >>>>>>> 2) A Value mapper is required to extract the foreign key.
> >>>>>>> table1 foreign key mapper: ( value => value.fk
<http://value.fk> )
> >>>>>>>
> >>>>>>> The mapper is applied to each element in table1, and a
new combined
> >>>>>>> key is
> >>>>>>> made:
> >>>>>>> table1 mapped: <A-a, (fk=A,bar=1)>, <A-b, (fk=A,bar=2)>,
<B-c,
> >>>>>>> (fk=B,bar=3)>
> >>>>>>>
> >>>>>>> 3) The rekeyed events are copartitioned with table2:
> >>>>>>>
> >>>>>>> a) Stream Thread with Partition 0:
> >>>>>>> RepartitionedTable1: <A-a, (fk=A,bar=1)>, <A-b,
(fk=A,bar=2)>
> >>>>>>> Table2: <A,X>
> >>>>>>>
> >>>>>>> b) Stream Thread with Partition 1:
> >>>>>>> RepartitionedTable1: <B-c, (fk=B,bar=3)>
> >>>>>>> Table2: <B,Y>
> >>>>>>>
> >>>>>>> 4) From here, they can be joined together locally by
applying the
> >>>>>>> joiner
> >>>>>>> function.
> >>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>> At this point, Jan's design and my design deviate. My
design goes
> on
> >>>>>>> to
> >>>>>>> repartition the data post-join and resolve out-of-order
arrival of
> >>>>>>> records,
> >>>>>>> finally returning the data keyed just the original key.
I do not
> >>>>>>> expose
> >>>>>>> the
> >>>>>>> CombinedKey or any of the internals outside of the
joinOnForeignKey
> >>>>>>> function. This does make for larger footprint, but it
removes all
> >>>>>>> agency
> >>>>>>> for resolving out-of-order arrivals and handling
CombinedKeys from
> the
> >>>>>>> user. I believe that this makes the function much easier
to use.
> >>>>>>>
> >>>>>>> Let me know if this helps resolve your questions, and
please feel
> >>>>>>> free to
> >>>>>>> add anything else on your mind.
> >>>>>>>
> >>>>>>> Thanks again,
> >>>>>>> Adam
> >>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>> On Tue, Sep 4, 2018 at 8:36 PM, Matthias J. Sax <
> >>>>>>> matth...@confluent.io <mailto:matth...@confluent.io>>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>> Hi,
> >>>>>>>
> >>>>>>>> I am just catching up on this thread. I did not read
everything so
> >>>>>>>> far,
> >>>>>>>> but want to share couple of initial thoughts:
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> Headers: I think there is a fundamental difference
between header
> >>>>>>>> usage
> >>>>>>>> in this KIP and KP-258. For 258, we add headers to
changelog topic
> >>>>>>>> that
> >>>>>>>> are owned by Kafka Streams and nobody else is supposed
to write
> into
> >>>>>>>> them. In fact, no user header are written into the
changelog topic
> >>>>>>>> and
> >>>>>>>> thus, there are not conflicts.
> >>>>>>>>
> >>>>>>>> Nevertheless, I don't see a big issue with using
headers within
> >>>>>>>> Streams.
> >>>>>>>> As long as we document it, we can have some "reserved"
header keys
> >>>>>>>> and
> >>>>>>>> users are not allowed to use when processing data with
Kafka
> Streams.
> >>>>>>>> IMHO, this should be ok.
> >>>>>>>>
> >>>>>>>> I think there is a safe way to avoid conflicts, since these
> headers
> >>>>>>>> are
> >>>>>>>>
> >>>>>>>>> only needed in internal topics (I think):
> >>>>>>>>> For internal and changelog topics, we can namespace
all headers:
> >>>>>>>>> * user-defined headers are namespaced as "external." +
headerKey
> >>>>>>>>> * internal headers are namespaced as "internal." +
headerKey
> >>>>>>>>>
> >>>>>>>>> While name spacing would be possible, it would require to
> >>>>>>>> deserialize
> >>>>>>>> user headers what implies a runtime overhead. I would
suggest to
> no
> >>>>>>>> namespace for now to avoid the overhead. If this becomes a
> problem in
> >>>>>>>> the future, we can still add name spacing later on.
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> My main concern about the design it the type of the
result KTable:
> >>>>>>>> If I
> >>>>>>>> understood the proposal correctly,
> >>>>>>>>
> >>>>>>>> KTable<K1,V1> table1 = ...
> >>>>>>>> KTable<K2,V2> table2 = ...
> >>>>>>>>
> >>>>>>>> KTable<K1,V3> joinedTable = table1.join(table2,...);
> >>>>>>>>
> >>>>>>>> implies that the `joinedTable` has the same key as the
left input
> >>>>>>>> table.
> >>>>>>>> IMHO, this does not work because if table2 contains
multiple rows
> >>>>>>>> that
> >>>>>>>> join with a record in table1 (what is the main purpose of a
> foreign
> >>>>>>>> key
> >>>>>>>> join), the result table would only contain a single
join result,
> but
> >>>>>>>> not
> >>>>>>>> multiple.
> >>>>>>>>
> >>>>>>>> Example:
> >>>>>>>>
> >>>>>>>> table1 input stream: <A,X>
> >>>>>>>> table2 input stream: <a,(A,1)>, <b,(A,2)>
> >>>>>>>>
> >>>>>>>> We use table2 value a foreign key to table1 key (ie,
"A" joins).
> If
> >>>>>>>> the
> >>>>>>>> result key is the same key as key of table1, this
implies that the
> >>>>>>>> result can either be <A, join(X,1)> or <A, join(X,2)>
but not
> both.
> >>>>>>>> Because the share the same key, whatever result record
we emit
> later,
> >>>>>>>> overwrite the previous result.
> >>>>>>>>
> >>>>>>>> This is the reason why Jan originally proposed to use a
> combination
> >>>>>>>> of
> >>>>>>>> both primary keys of the input tables as key of the
output table.
> >>>>>>>> This
> >>>>>>>> makes the keys of the output table unique and we can
store both in
> >>>>>>>> the
> >>>>>>>> output table:
> >>>>>>>>
> >>>>>>>> Result would be <A-a, join(X,1)>, <A-b, join(X,2)>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> Thoughts?
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> -Matthias
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On 9/4/18 1:36 PM, Jan Filipiak wrote:
> >>>>>>>>
> >>>>>>>> Just on remark here.
> >>>>>>>>> The high-watermark could be disregarded. The decision
about the
> >>>>>>>>> forward
> >>>>>>>>> depends on the size of the aggregated map.
> >>>>>>>>> Only 1 element long maps would be unpacked and
forwarded. 0
> element
> >>>>>>>>> maps
> >>>>>>>>> would be published as delete. Any other count
> >>>>>>>>> of map entries is in "waiting for correct deletes to
> arrive"-state.
> >>>>>>>>>
> >>>>>>>>> On 04.09.2018 21:29, Adam Bellemare wrote:
> >>>>>>>>>
> >>>>>>>>> It does look like I could replace the second
repartition store
> and
> >>>>>>>>>> highwater store with a groupBy and reduce. However,
it looks
> like
> >>>>>>>>>> I
> >>>>>>>>>> would
> >>>>>>>>>> still need to store the highwater value within the
materialized
> >>>>>>>>>> store,
> >>>>>>>>>>
> >>>>>>>>>> to
> >>>>>>>>> compare the arrival of out-of-order records (assuming my
> >>>>>>>>> understanding
> >>>>>>>>> of
> >>>>>>>>> THIS is correct...). This in effect is the same as the
design I
> have
> >>>>>>>>> now,
> >>>>>>>>> just with the two tables merged together.
> >>>>>>>>>
>
>
