Full Outer Join

[phisinger]

Hello all,

thank you for this great application. It seems that it is exactly what I searched for for a long time.

Nevertheless, I have a question: I need to implement a (SQL-like) Full Outer Join of multiple streams in my Flink Job.
The background is, I want to join messages from different streams if a join is possible; if there isn't a match of the keys in a given time window, I want to output the fields of the input streams in separate messages (the fields from the other stream should be null).

I've seen that there is a node called "union-memo", which is like an Inner Join. Second, there is a "single side join" that could be combined with a union node, if I got the functionality right.

Is there a node that delivers a Full Outer Join? Otherwise, can I use the single side join and a union node to get a full outer join in the end?
I'm looking forward to your thoughts.

[arkadius]

Hi @phisinger

Thank you for warm word.

Feature that you are writing about (Full Outer Join component) looks like something that would be a nice complementation of set of stateful components thats are available in Nussknacker distribution.

Currently component that works quite similar is union-memo. The only difference that I see is that union-memo has state which is "sustained" by inputs from each sides. It means that for sequence:

• T0: event1 from A side
• T0+1min: event2 from B side
• T0+2min: event3 from A side
• T0+4min: event4 from A side

and window (state timeout) with span 2 minutes , after event4 will be collected event with values:

{
A: event4
B: event2
}

For Outer Join you rather expect that B will null in this case.

Am I right that this difference is crucial for your usage and union-memo won't be a good replacement for that?

On the other hand, single side join is a "real" outer join but asymmetric. You can try to make something like two splits for each sources and after that two single side joins but with reversed MAIN branchType and then normal union.

You can also try to write your own component based on single side join and union-memo implementations. If you have basic knowledge of Flink, it won't be hard. The implementation of components is represented as Flink's DSL. We have a Developer Guide: https://nussknacker.io/documentation/docs/developers_guide/Basics (which is still in WIP phase, but we can help you if you ocurres some troubles using our components API). We also prepared projects showing how to write your own sample component from scratch: https://github.com/TouK/nussknacker-sample-components

Let me know if I help you.

[phisinger]

Hey @arkadius,

thanks for this fast and comprehensive answer.
After your explanation, I think I should try it again with union-memo.

But I have got some other questions:

• Messages from A side and B side are only merged when their keys are matching, right?

If the sequence is like the following one with time window = 2 min:

1. T0: event1(key=X) from A side
2. T0+3min: event2(key=X) from B side
3. T0+4min: event3(key=Y) from A side
4. T0+6min: event4(key=Z) from A side

What would be the first output message, or would there be an output like {A: event1(key=X), B: NULL} at T0+2min? Or would there even be an output?
What would be the output after event3 is collected? Would it be {A: event3(key=Y), B: NULL}?

I know that these are rather strange cases and questions. For our use case, it's more important that an event is outputted"fast" (considering the time window) than that it is merged with others. We rather output a message that is half NULL than wait (maybe forever for the matching key) and merge an event with another event.

Thank you very much again for your help! I will otherwise try it with the single-side joins or writing my own node.

[arkadius]

Messages from A side and B side are only merged when their keys are matching, right?
Yes, it is correct. Messages for other keys are "not visible" for given key

What would be the first output message, or would there be an output like {A: event1(key=X), B: NULL} at T0+2min? Or would there even be an output?

"union-memo" like just "union" ensures that for every incoming message is always generated exact one outgoing message. So outgoing messages will look like:

1. T0: A: event1(key=X), B : null
2. T0+3min: A: null, B: event2(key=X)
3. T0+4min: A: event3(key=Y), B: null
4. T0+6min: A: event4(key=Z), B: null
   The "other side" is always null in the example above because delay between messages with the same key (1. and 2.) was greater than
   window span.

Time of emitting messages is the same as time of incoming messages. This component is rather simple - it doesn't wait in some window, it just "keep state" for time of this window but react on every event instantly. It is the thing that probably be changed or configurable in some more high level component like "Full outer join" because it could cause some run condition in case of desynchronized branches.

Summarizing - this component is "fast". The problem is if it is not "too fast" for your cases?

[phisinger]

Although this is a very good answer, take a look at the new full-outer-join component mentioned below

[arkadius]

Hi @phisinger . I'm back after a longer delay. Have you find a work around for your issue? Meantime we've developed full outer join component which will work (I hope) as you expected. You can read about its in our development version of documentation: https://nussknacker.io/documentation/docs/next/scenarios_authoring/AggregatesInTimeWindows/#full-outer-join . Implementation and test cases you can find in MR: https://github.com/TouK/nussknacker/pull/3075/files . This change will be released in 1.5 version. We have a plan to release it around 2022-08-16. We will be glad to hear if this component fulfill your requirements.

[phisinger]

Hey @arkadius,
thank you very much for reaching out to me again. Unfortunately, I don't have the ressources to test the component at the moment. But it seems very good. I also have the impression that your updated the supported aggregators (or at least the docu). Looks very intuitiv for me now. As soon as I can and need to use nussknacker again I will share my experience with the new component. Thank you!