Hi Eva, Correct me If i'm wrong. You have an unbounded Task stream and you want to enrich the User info to the task event. Meanwhile, the User table is also changing by the time, so you basically want that when task event comes, join the latest data of User table and emit the results. Even if the User table changes again, you don't want to re-trigger the join operation which happened before and already emitted, right?
Best, Kurt On Fri, Dec 20, 2019 at 12:33 AM Timo Walther <twal...@apache.org> wrote: > Hi Eva, > > I'm not 100% sure if your use case can be solved with SQL. JOIN in SQL > always joins an incoming record with all previous arrived records. Maybe > Jark in CC has some idea? > > It might make sense to use the DataStream API instead with a connect() > and CoProcessFunction where you can simply put the latest row into state > and perform the joining and emission of a new row when required. > > Regards, > Timo > > > On 18.12.19 23:44, Eva Eva wrote: > > Hi Team, > > > > I'm trying Flink for the first time and encountered an issue that I > > would like to discuss and understand if there is a way to achieve my use > > case with Flink. > > > > *Use case:* I need to perform unbounded stream joins on multiple data > > streams by listening to different Kafka topics. I have a scenario to > > join a column in a table with multiple columns in another table by > > avoiding duplicate joins. The main concern is that I'm not able to avoid > > duplicate joins. > > > > *Issue: *Given the nature of data, it is possible to have updates over > > time, sent as new messages since Kafka is immutable. For a given key I > > would like to perform join only on the latest message, whereas currently > > Flink performs join against all messages with the key (this is what I'm > > calling as duplicate joins issue). > > Example: Say I have two Kafka streams "User" and "Task". And I want to > > join "User" with multiple columns in "Task". > > Join "UserID" in "User" with "PrimaryAssignee", "SecondaryAssignee" and > > "Manager" in "Task". > > > > Assuming I created and registered DataStreams. > > Below is my query: > > > > SELECT * FROM Task t > > LEFT JOIN User ua ON t.PrimaryAssignee = ua.UserID > > LEFT JOIN User ub ON t.SecondaryAssignee = ub.UserID > > LEFT JOIN User uc ON t.Manager = uc.UserID > > > > Say I have 5 different messages in Kafka with UserID=1000, I don't want > > to perform 5 joins instead I want to perform join with the only latest > > message with UserID=1000. Is there any way to achieve this without using > > Temporal Table Functions? > > > > *I cannot use Temporal Table Functions because of below reasons:* > > 1. I need to trigger JOIN operation for every new message in Kafka. > > Whereas new messages in Temporal Table don't trigger JOIN operation. > > 2. I need to perform LEFT OUTER JOINS, whereas Temporal Table can only > > be used for INNER JOINS > > 3. From what I understand, JOIN in Temporal Table can only be performed > > using Primary key, so I won't be able to Join more than one key. > > > > > > Could someone please help me with this? Please let me know if any of the > > information is not clear or need more details. > > > > If this is not the correct email id, could you please point me to the > > correct one. > > > > > > Thanks in advance! > >
