join performance

[Henry Cai <hc...@pinterest.com>]

For the below standard stream/stream join, does flink store the results of
stream 1 and stream 2 into state store for the current hour and at the end
of the hour window it will fire the window by iterating through all stored
elements in the state store to find join matches?

My concern is during most of the time in the hour, the output (assuming the
output is going to another stream) will be idle and on each hour mark there
will be huge outputs of joined records emitted, any way to make it more
gradual?


dataStream.join(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .apply (new JoinFunction () {...});

Re: join performance

[Aljoscha Krettek <al...@apache.org>]

Hi Henry,
yes, with early firings you would have the problem of duplicate emission.
I'm afraid I don't have a solution for that right now.

For the "another question" I think you are right that this would be session
windowing. Please have a look at this blog post that I wrote recently:
http://data-artisans.com/session-windowing-in-flink/. And please get back
to us if you have more questions or feedback.

Cheers,
Aljoscha

On Fri, 29 Apr 2016 at 19:18 Henry Cai <hc...@pinterest.com> wrote:

> So is the window defined as hour-window or second-window?
>
> If I am using hour-window, I guess I need to modify the trigger to fire
> early (e.g. every minute)?  But I don't want to repeatedly emit the same
> joined records for every minute (i.e. on 2nd minute, I only want to emit
> the changes introduced by new coming records between 1st and 2nd minute)
>
> If I am using second-window, I wasn't sure why the record will still be
> put into the correct window based on hour gap?
>
> Another question is on which type of window, I need to match record a from
> stream a to record b in stream b if abs(a.time - b.time) < 1-hour, so it's
> not really a tumbling window on absolute wall clock, is this a session
> window?
>
> On Fri, Apr 29, 2016 at 4:36 AM, Aljoscha Krettek <al...@apache.org>
> wrote:
>
>> Hi,
>> you are right, everything will be emitted in a huge burst at the end of
>> the hour. If you want to experiment a bit you can write a custom Trigger
>> based on EventTimeTrigger that will delay firing of windows. You would
>> change onEventTime() to not fire but instead register a processing-time
>> timer at a random point in the future. Then, in onProcessingTime() you
>> would trigger the actual window processing. Elements will still be put into
>> the correct windows based on event time, just the firing of the windows
>> will change by doing this.
>>
>> Cheers,
>> Aljoscha
>>
>> On Fri, 29 Apr 2016 at 08:53 Henry Cai <hc...@pinterest.com> wrote:
>>
>>> But the join requirement is to match the records from two streams
>>> occurring within one hour (besides the normal join key condition), if I use
>>> the second join window, those records wouldn't be in the same window any
>>> more.
>>>
>>>
>>>
>>> On Thu, Apr 28, 2016 at 11:47 PM, Ashutosh Kumar <
>>> kmr.ashutosh16@gmail.com> wrote:
>>>
>>>> Time unit can be in seconds as well. Is there specific need to get
>>>> bursts hourly?
>>>>
>>>> On Fri, Apr 29, 2016 at 11:48 AM, Henry Cai <hc...@pinterest.com> wrote:
>>>>
>>>>> For the below standard stream/stream join, does flink store the
>>>>> results of stream 1 and stream 2 into state store for the current hour and
>>>>> at the end of the hour window it will fire the window by iterating through
>>>>> all stored elements in the state store to find join matches?
>>>>>
>>>>> My concern is during most of the time in the hour, the output
>>>>> (assuming the output is going to another stream) will be idle and on each
>>>>> hour mark there will be huge outputs of joined records emitted, any way to
>>>>> make it more gradual?
>>>>>
>>>>>
>>>>> dataStream.join(otherStream)
>>>>>     .where(0).equalTo(1)
>>>>>     .window(TumblingEventTimeWindows.of(Time.hours(1)))
>>>>>     .apply (new JoinFunction () {...});
>>>>>
>>>>>
>>>>
>>>
>

Re: join performance

[Henry Cai <hc...@pinterest.com>]

So is the window defined as hour-window or second-window?

If I am using hour-window, I guess I need to modify the trigger to fire
early (e.g. every minute)?  But I don't want to repeatedly emit the same
joined records for every minute (i.e. on 2nd minute, I only want to emit
the changes introduced by new coming records between 1st and 2nd minute)

If I am using second-window, I wasn't sure why the record will still be put
into the correct window based on hour gap?

Another question is on which type of window, I need to match record a from
stream a to record b in stream b if abs(a.time - b.time) < 1-hour, so it's
not really a tumbling window on absolute wall clock, is this a session
window?

On Fri, Apr 29, 2016 at 4:36 AM, Aljoscha Krettek <al...@apache.org>
wrote:

> Hi,
> you are right, everything will be emitted in a huge burst at the end of
> the hour. If you want to experiment a bit you can write a custom Trigger
> based on EventTimeTrigger that will delay firing of windows. You would
> change onEventTime() to not fire but instead register a processing-time
> timer at a random point in the future. Then, in onProcessingTime() you
> would trigger the actual window processing. Elements will still be put into
> the correct windows based on event time, just the firing of the windows
> will change by doing this.
>
> Cheers,
> Aljoscha
>
> On Fri, 29 Apr 2016 at 08:53 Henry Cai <hc...@pinterest.com> wrote:
>
>> But the join requirement is to match the records from two streams
>> occurring within one hour (besides the normal join key condition), if I use
>> the second join window, those records wouldn't be in the same window any
>> more.
>>
>>
>>
>> On Thu, Apr 28, 2016 at 11:47 PM, Ashutosh Kumar <
>> kmr.ashutosh16@gmail.com> wrote:
>>
>>> Time unit can be in seconds as well. Is there specific need to get
>>> bursts hourly?
>>>
>>> On Fri, Apr 29, 2016 at 11:48 AM, Henry Cai <hc...@pinterest.com> wrote:
>>>
>>>> For the below standard stream/stream join, does flink store the results
>>>> of stream 1 and stream 2 into state store for the current hour and at the
>>>> end of the hour window it will fire the window by iterating through all
>>>> stored elements in the state store to find join matches?
>>>>
>>>> My concern is during most of the time in the hour, the output (assuming
>>>> the output is going to another stream) will be idle and on each hour mark
>>>> there will be huge outputs of joined records emitted, any way to make it
>>>> more gradual?
>>>>
>>>>
>>>> dataStream.join(otherStream)
>>>>     .where(0).equalTo(1)
>>>>     .window(TumblingEventTimeWindows.of(Time.hours(1)))
>>>>     .apply (new JoinFunction () {...});
>>>>
>>>>
>>>
>>

Re: join performance

[Aljoscha Krettek <al...@apache.org>]

Hi,
you are right, everything will be emitted in a huge burst at the end of the
hour. If you want to experiment a bit you can write a custom Trigger based
on EventTimeTrigger that will delay firing of windows. You would change
onEventTime() to not fire but instead register a processing-time timer at a
random point in the future. Then, in onProcessingTime() you would trigger
the actual window processing. Elements will still be put into the correct
windows based on event time, just the firing of the windows will change by
doing this.

Cheers,
Aljoscha

On Fri, 29 Apr 2016 at 08:53 Henry Cai <hc...@pinterest.com> wrote:

> But the join requirement is to match the records from two streams
> occurring within one hour (besides the normal join key condition), if I use
> the second join window, those records wouldn't be in the same window any
> more.
>
>
>
> On Thu, Apr 28, 2016 at 11:47 PM, Ashutosh Kumar <kmr.ashutosh16@gmail.com
> > wrote:
>
>> Time unit can be in seconds as well. Is there specific need to get bursts
>> hourly?
>>
>> On Fri, Apr 29, 2016 at 11:48 AM, Henry Cai <hc...@pinterest.com> wrote:
>>
>>> For the below standard stream/stream join, does flink store the results
>>> of stream 1 and stream 2 into state store for the current hour and at the
>>> end of the hour window it will fire the window by iterating through all
>>> stored elements in the state store to find join matches?
>>>
>>> My concern is during most of the time in the hour, the output (assuming
>>> the output is going to another stream) will be idle and on each hour mark
>>> there will be huge outputs of joined records emitted, any way to make it
>>> more gradual?
>>>
>>>
>>> dataStream.join(otherStream)
>>>     .where(0).equalTo(1)
>>>     .window(TumblingEventTimeWindows.of(Time.hours(1)))
>>>     .apply (new JoinFunction () {...});
>>>
>>>
>>
>

Re: join performance

[Henry Cai <hc...@pinterest.com>]

But the join requirement is to match the records from two streams occurring
within one hour (besides the normal join key condition), if I use the
second join window, those records wouldn't be in the same window any more.



On Thu, Apr 28, 2016 at 11:47 PM, Ashutosh Kumar <km...@gmail.com>
wrote:

> Time unit can be in seconds as well. Is there specific need to get bursts
> hourly?
>
> On Fri, Apr 29, 2016 at 11:48 AM, Henry Cai <hc...@pinterest.com> wrote:
>
>> For the below standard stream/stream join, does flink store the results
>> of stream 1 and stream 2 into state store for the current hour and at the
>> end of the hour window it will fire the window by iterating through all
>> stored elements in the state store to find join matches?
>>
>> My concern is during most of the time in the hour, the output (assuming
>> the output is going to another stream) will be idle and on each hour mark
>> there will be huge outputs of joined records emitted, any way to make it
>> more gradual?
>>
>>
>> dataStream.join(otherStream)
>>     .where(0).equalTo(1)
>>     .window(TumblingEventTimeWindows.of(Time.hours(1)))
>>     .apply (new JoinFunction () {...});
>>
>>
>

Re: join performance

[Ashutosh Kumar <km...@gmail.com>]

Time unit can be in seconds as well. Is there specific need to get bursts
hourly?

On Fri, Apr 29, 2016 at 11:48 AM, Henry Cai <hc...@pinterest.com> wrote:

> For the below standard stream/stream join, does flink store the results of
> stream 1 and stream 2 into state store for the current hour and at the end
> of the hour window it will fire the window by iterating through all stored
> elements in the state store to find join matches?
>
> My concern is during most of the time in the hour, the output (assuming
> the output is going to another stream) will be idle and on each hour mark
> there will be huge outputs of joined records emitted, any way to make it
> more gradual?
>
>
> dataStream.join(otherStream)
>     .where(0).equalTo(1)
>     .window(TumblingEventTimeWindows.of(Time.hours(1)))
>     .apply (new JoinFunction () {...});
>
>