Hi Dongwon,
You are of course right! We need to decrement the counter
when the window is closed.
The idea of using Trigger.clear() (the clean up method is
called clear() instead of onClose()) method is great!
It will be called when the window is closed but also when it
is merged.
So, I think you are right and we only need to increment the
counter in Trigger.onElement() and decrement in Trigger.clear().
I'm not 100% sure, but I doubt that metrics can checkpointed.
Chesnay (in CC) would know that.
Not sure what would be the best approach if you need a fault
tolerant solution.
Best, Fabian
2018-06-19 16:38 GMT+02:00 Dongwon Kim <eastcirc...@gmail.com
<mailto:eastcirc...@gmail.com>>:
Hi Fabian,
Thanks a lot for your reply.
Do you need to number of active session windows as a
DataStream or would you like to have it as a metric
that you can expose.
I possible, I would recommend to expose it as a
metric because they are usually easier to collect.
I want to have it as a metric and it doesn't look
difficult thanks to the metric system exposed by
TriggerContext.
In order to track how many session windows exist, we
would need to increment a counter by one when a new
window is created (or an element is assigned to a
window, which is equivalent for session windows)
I agree with you that we need to increment a counter when
Trigger.onElement() is called due to the characteristic
of session windows.
and decrement the counter when windows are merged by
the number of merged windows minus one.
You decrement the counter when windows are merged, but I
think we need to decrement the counter when a window is
expired as well.
However, decrementing the counter is difficult.
Although the Trigger.onMerge() method is called, it
does not know how many windows were merged (which is
done by the WindowAssigner) and only sees the merged
window.
We assume that timestamps of records from a user are in
ascending order, so only one window is closed at a time
which simplifies the problem of how to decrement the counter.
Nevertheless, I think I need to decrement the counter in
Trigger.onClose(), not Trigger.onMerge().
By doing that in Trigger.onClose(), we can take care of
both cases: when a window is merged and when a window is
expired.
How do you think about it?
The reason I mention state is to calculate the exact
number of active sessions even after my Flink application
is restarted from checkpoints or savepoints.
If we restore from a savepoint and the counter is
initialized to 0, we'll see an incorrect value from a
dashboard.
This is the biggest concern of mine at this point.
Best,
- Dongwon
On Tue, Jun 19, 2018 at 7:14 PM, Fabian Hueske
<fhue...@gmail.com <mailto:fhue...@gmail.com>> wrote:
Hi Dongwon,
Do you need to number of active session windows as a
DataStream or would you like to have it as a metric
that you can expose.
I possible, I would recommend to expose it as a
metric because they are usually easier to collect.
SessionWindows work internally as follows:
- every new record is added to a new window that
starts at the timestamp of the record and ends at
timestamp + gap size. When a record is added to a
window, Trigger.onElement() is called.
- after a window was created, the session window
assigner tries to merge window with overlapping
ranges. When windows are merged, Trigger.onMerge() is
called.
In order to track how many session windows exist, we
would need to increment a counter by one when a new
window is created (or an element is assigned to a
window, which is equivalent for session windows) and
decrement the counter when windows are merged by the
number of merged windows minus one.
Incrementing the counter is rather easy and can be
done in Trigger.onElement(), either by using state or
a Counter metric (Triggers have access to the metric
system).
However, decrementing the counter is difficult.
Although the Trigger.onMerge() method is called, it
does not know how many windows were merged (which is
done by the WindowAssigner) and only sees the merged
window. There might be a way to maintain state in a
Trigger that allows to infer how many windows were
merged.
Best, Fabian
2018-06-16 16:39 GMT+02:00 Dongwon Kim
<eastcirc...@gmail.com <mailto:eastcirc...@gmail.com>>:
Hi Fabian,
I'm still eager to expose # of active sessions as
a key metric of our service but I haven’t figured
it out yet.
First of all, I want to ask you some questions
regarding your suggestion.
You could implement a Trigger that fires when a
new window is created and when the window is
closed. A ProcessWindowFunction would emit a +1
if the window was created and a -1 when the
window is closes.
Session windows are a bit special, because you
also need to handle the case of merging windows,
i.e., two opened windows can be merged and only
one (the merged) window is closed. So would need
to emit a -2 if a merged window was closes
(assuming only two windows were merged).
Q1)
How to fire when a new window is created and when
the window is closed?
AFAIK, we can return TriggerResult only through
the three functions: onElement, onEventTime, and
onProcessingTime.
Q2)
Firing is to emit elements in windows down to the
window function, not emitting values like +1, -1
and -2 which are not in windows.
Or do I miss something that you meant?
In order to do that, you'd need to carry the
merging information forward. The Trigger.onMerge
method cannot trigger the window function, but
it could store the merging information in state
that is later accessed.
Q3)
I didn't understand what you mean at all. What do
you mean by carrying the merging information?
Besides your suggestion, I implemented a custom
trigger which is almost the same as
EventTimeTrigger except the followings:
- it maintains a variable to count sessions in an
instance of a window operator
- it increases the variable by 1 when onElement
is invoked
- it decreases the variable by 1 when onClose is
invoked
Considering the logic of Flink’s session window,
it correctly counts sessions in an instance of a
window operator.
As you might have already noticed, this approach
has a critical problem: there's no way to
maintain an operator state inside a trigger.
TriggerContext only allows to interact with state
that is scoped to the window and the key of the
current trigger invocation (as shown in
Trigger#TriggerContext)
Now I've come to a conclusion that it might not
be possible using DataStream API.
Otherwise, do I need to think in a totally
different way to achieve the goal?
Best,
- Dongwon
2018. 2. 20. 오후 6:53, Fabian Hueske
<fhue...@gmail.com <mailto:fhue...@gmail.com>> 작성:
Hi Dongwon Kim,
That's an interesting question.
I don't have a solution blueprint for you, but a
few ideas that should help to solve the problem.
I would start with a separate job first and
later try to integrate it with the other job.
You could implement a Trigger that fires when a
new window is created and when the window is
closed. A ProcessWindowFunction would emit a +1
if the window was created and a -1 when the
window is closes.
Session windows are a bit special, because you
also need to handle the case of merging windows,
i.e., two opened windows can be merged and only
one (the merged) window is closed. So would need
to emit a -2 if a merged window was closes
(assuming only two windows were merged).
In order to do that, you'd need to carry the
merging information forward. The Trigger.onMerge
method cannot trigger the window function, but
it could store the merging information in state
that is later accessed.
Hope this helps,
Fabian
2018-02-20 9:54 GMT+01:00 Dongwon Kim
<eastcirc...@gmail.com
<mailto:eastcirc...@gmail.com>>:
Hi,
It could be a totally stupid question but I
currently have no idea how to get the number
of active session windows from a running job.
Our traffic trajectory application (which
handles up to 10,000 tps) uses event-time
session window on KeyedStream (keyed by userID).
Should I write another Flink job for the
purpose?
Cheers,
Dongwon Kim
