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 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> 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>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 originalpartitions 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> 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 MatthiasThank you for your feedback, I do appreciate it! While name spacing would be possible, it would require to deserializeuser 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 ) 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> 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 areonly 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 todeserialize 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 andhighwater store with a groupBy and reduce. However, it looks like I would still need to store the highwater value within the materialized store, tocompare 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.
