[FLINK-3109]Join two streams with two different buffer time -- Java i…

[wangyangjun]

Java implementation of jira FLINK-3109

[tillrohrmann]

System.out is not good here. Maybe using logger if you want to log something.

[wangyangjun]

Thanks for your suggestion.

[aljoscha]

Hi @wangyangjun,
sorry for the long wait but I think we can get this PR in after some modifications. I'd like to change the API a bit to separate it from the other Join implementation since that class is already quite crowded. What I would propose is to add a method timeJoin() on DataStream and a new class TimeJoinedStreams that is similar to JoinedStreams but specific to the two-buffer time join.

Could you also please add support for the Scala API, we try to keep the two APIs in sync. If you need help with that, please let me know.

[wangyangjun]

Hello, @aljoscha ,
Thanks for your comment and suggestion. I will update the API. During my benchmark tests of timeJoin, the feature of this implementation is quite good, but the performance is bad. I will reimplement it with Guava.

Yes, I could add support for the Scala API.

Thanks

[wangyangjun]

Hello @aljoscha ,
As I mentioned in last comment, I will reimplement it with Guava CacheBuilder. One question is about Flink self-memory management, does CacheBuilder get memory from Flink or JVM directly? As I understand, HeapWindowBuffer allocates memory from Flink. Is there any data structure like cachedMap in Flink?

[StephanEwen]

Concerning the data management: @aljoscha and me are currently heavily reworking that.
All window operations need to go onto the "state" interfaces. Before we merge this one, we should also do that, so please do not spend much time on optimizing how the buffers for the two inputs are implemented.

The interfaces for that will go into the code in a few days (they are in this pull request: #1562)

For now, I would focus on the API and we look into the buffers in a few days.

BTW: how exactly the buffered data is held (managed memory, external databases, etc) depends on the "state backend" of the job. Memory behavior can be changed that way and the operators need not worry about that.

[StephanEwen]

@wangyangjun We actually merged all the changes concerning the state abstraction.

To make this window join work seamlessly on Flink's state backends (memory, or key/value stores, managed memory, ...) you would need to implement it against the key/value state. That means that whenever you store data in the operator, the data should go into the partitioned state that you can obtain from the AbstractStreamOperator or the RuntimeContext.

I think that for this window operator, the ListState is a good choice, where you can add values to a key and retrieve the list as a whole once the windows are evaluated.

Please write back if you need some more pointers on the state abstraction.

[StephanEwen]

I just saw that you updated this pull request (actually a few weeks ago already)

A lot of it looks very good, some things we need to check a bit deeper (like how triggers actually behave on the two separate windows, how windows are matched).

Can you give a high level summary of how this should behave?
Especially given that you allow for custom triggers and window assigners here, how are windows matched against each other (to determine that their elements should be joined/co-grouped).
For tumbling time windows, the behavior is well defined and like discussed in the JIRA issue, but for generic windows and triggers, how is it defined?

[wangyangjun]

Hi StephanEwen, sorry for replying this late.

This will join two window streams as long as slide size of these two windows are equal. It can be two SlidingTimeWindows, one SlidingTimeWindow and one TumblingTimeWindow. If slide size of windows are not equal, an exception will be threw.

It doesn't support generic windows and triggers yet. Only TimeWindow is allowed in the API.

[aljoscha]

I'm closing this as "Abandoned", since there is no more activity and the code base has moved on quite a bit.