[
https://issues.apache.org/jira/browse/SAMZA-552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14350039#comment-14350039
]
Jay Kreps commented on SAMZA-552:
---------------------------------
Hey Yi, two quick questions where I'm not sure if I follow:
1. Why is there a concept of "window close" at all? Once you allow for
corrections you are basically treating a window as mutable state and issuing
corrections. So why not just view the entire computation of the window as a
running stream of updates from beginning to end. There is no real difference
between updates before the window close and after the window close. Can't we
just get rid of the whole concept of window close? Whenever an update comes in
it updates something and that logs out the new value, this is true whether you
are currently in the window or whether (according to some clock) the window has
"closed". All the complexity of the millwheel model can just go away. The only
real "close" concept that exists is the retention window...once you get updates
to data outside your retention window you have to toss it out. I agree that the
problem of what clock to base the retention window on remains. Basically if I
understand what you are saying I think you are still trying for a model where
we attempt to issue a single canonical output for a window, and then correct it
only if late data arrives. But what I am advocating is just using the existing
state storage mechanism for the aggregate and having the changelog for that be
the result of the query. Since you have thought this through more I think the
question I was hoping to see answered is whether we can just make the hard
assumption that "aggregates" == "key value store changelog" with no additional
support needed?
I think this matters because if you try to avoid output until the end of the
window, you will need to have some backup for the state in the key-value store,
so you will still need a changelog. But that means having both a changelog and
an output stream with almost identical contents, right?
2. In general the materialized result in the k/v store will be the aggregate
not the raw rows, right? I.e. in your sql example we would be storing the
mapping hour=>count not hour=>{events}. I think this is what you are saying,
but I'm not sure. So in the counting case there is no real "recomputation" it
is just a matter of processing the new row and outputing the final value
(count+1). I think this is often true, but maybe not always? Maybe joins and
non-incremental aggregates like median are counter-examples? Perhaps this is
just a sort of optimization done in the query planning layer to push down the
aggregation all the way to make the aggregate itself incremental even though
the value is in some sense equivalent to storing all the rows and recomputing
from scratch on each update?
> 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)
> Attachments: DESIGN-SAMZA-552-3.md, DESIGN-SAMZA-552-3.pdf
>
>
> 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)
