[
https://issues.apache.org/jira/browse/SAMZA-552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14339783#comment-14339783
]
Jay Kreps commented on SAMZA-552:
---------------------------------
Yeah one thing it reminded me of was an idea I've had a couple times with
respect to windowing, basically just treating everything as updatable as a way
to handle late data. Once you get rid of the idea that grouping over a window
produces only a single output for the group key the whole problem of late data
is kind of solved.
The actual time scheme in MillWheel has always struck me as a bit complex. But
this talk reminded me of the same idea, which is anything on event time windows
needs to be updateable. I'm not sure you even need their whole
marker/trigger/watermark scheme.
Consider a case where you are doing
select trunc_to_minute(time) min, user_id, count(1)
from page_views
group by trunc_to_minute(time), user_id;
Imagine that you are using our key-value store to capture count data as it
occurs by doing
put(user_id + minute, get(user_id + minute) + 1)
Say you have input
Time, User
12:01:01, 1
12:01:03, 1
12:01:05, 2
12:01:09, 1
In the absence of any caching the changelog stream would basically capture
something like
12:01, 1, 1
12:01, 1, 2
12:01, 2, 1
12:01. 1, 3
In other words, the stream of updates.
The whole windowing scheme they propose is just a way of suppressing some of
those updates so that you give just one count if there is no late data.
But effectively you can view the caching layer in the key value store as doing
a similar thing--attempting to supress duplicate updates until the commit point.
If the downstream thing needs to know to treat the data like a table that is
updated, which it will have to do if you ever admit the possibility of late
data, then it will have to handle updates. So in that case the pruning of
duplicate updates is really just a performance optimization.
The configuration you have to provide is what TTL to have in RocksDB because at
some point you are just going to toss the old counts and at that point you
really can't take anything later, but that could be days.
> Tuple or time window semantics in physical operator
> ---------------------------------------------------
>
> Key: SAMZA-552
> URL: https://issues.apache.org/jira/browse/SAMZA-552
> Project: Samza
> Issue Type: Sub-task
> Components: sql
> Affects Versions: 0.9.0
> Reporter: Yi Pan (Data Infrastructure)
> Assignee: Yi Pan (Data Infrastructure)
>
> The discussion is based on how to support tuple and/or time based window
> operators in Samza physical operator layer.
> Here are the few observations:
> # Tuple represents the “physical ordering” of events while time-based window
> has semantic meanings to users
> # Total ordering between tuples are possible within Samza/Kafka given a
> deterministic MessageSelector on all input streams and offsets within each
> stream
> # No matter whether tuple or time is used to measure the window size, the
> window termination condition is needed to close a window to avoid the job to
> be wedged forever
> The following questions have to be answered to fully implement a window
> operator:
> # how to determine that a window is closed and no new tuples will be added?
> ## For tuple based, how do we close the window if messages do not come or get
> delayed?
> ## For time based, how do we close the window if
> ### the messages are not strictly in order w/ the time?
> ### the message w/ timestamp greater than the window boundary does not come
> or gets delayed?
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
