[
https://issues.apache.org/jira/browse/SAMZA-552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14349328#comment-14349328
]
Yi Pan (Data Infrastructure) commented on SAMZA-552:
----------------------------------------------------
Just to document some of the discussion items on the MillWheel model:
We did not think about handling late arrivals as a "correction" to the earlier
results. If we follow that model, the late arrival problem can be addressed in
the following:
a) Each window will have a closing policy: it would either be wall-clock based
timeout, or the arrival of messages indicating that we have received all
messages in the corresponding event time window
b) Each window also keeps all the past messages it receives in the past
windows, up to a large retention size that covers all possible late arrivals
c) When a window's closing policy is satisfied, the window operator always
emits the current window results
d) When a late arrival message came, the window operator will re-emit the past
window results to correct the previous window results
Looking at a windowed aggregation example:
{code}
SELECT FLOOR(event_time TO HOUR) as time, COUNT(*) as counter FROM Stream GROUP
BY FLOOR(event_time TO HOUR)
{code}
In this example, the aggregation for the counter for window from 10:00-10:59
will have a "wrong" value when the window is closed by an arrival of message w/
11:00 timestamp, but will be corrected later by a late arrival of another
message in the time window from 10:00-10:59. I.e. if we keep all the previous
window states, late arrival messages will simply trigger a re-computation of
the aggregated counter for the window 10:00-10:59 and overwrite the previous
result. In this model, the final result is always correct, as long as the late
arrivals is within the large retention size.
It seems that the followings are more reasonable:
1) Window operator will have a full buffered state of the stream similar to a
time-varying materialized view over the retention size
2) Window size and termination (i.e. sliding/tumbling/hopping windows) will now
determine when we emit window results (i.e. new messages/updates to the current
window) to the downstream operator s.t. the operators can calculate result in
time
3) Late arrivals will be sent to the downstream operator and triggers a
re-computation of the past result based on the full buffered state
4) Optionally, the downstream operator can decide to suppress the output for a
while to allow late arrivals to correct the result before sending the output.
In the above model, the window operator becomes a system feature, or an
implementation of "StreamScan" in Calcite's term. And we do not need specific
language support for the window semantics, with a default time window operator
implementation that serves as a "StreamScan". All window definition in the
query language now only dictates the semantic meaning of aggregation and join
on top of the physical window operator which provides: a) a varying/growing
materialized view; b) a driver that tells the aggregation/join to
compute/re-compute results on-top-of the materialized view.
I will update the design doc to reflect the above thoughts.
> 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)
