[
https://issues.apache.org/jira/browse/SAMZA-552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14340604#comment-14340604
]
Chris Riccomini commented on SAMZA-552:
---------------------------------------
One thing that I found slightly odd about the description of late data in the
talk was that it seems that their solution only works for some use cases. For
example, if you are doing a LEFT JOIN between two streams "a" and "b". At some
point, you will emit the results. If a message has arrived for "a", but not for
"b", then you'll emit "a", NULL. If (within the 2 week time span described in
the talk) the corresponding "b" arrives, you'd then emit "a","b". Now your
stream has:
"a", NULL
"a", "b"
Now suppose you had two counters: "hits" and "misses". You increment hit every
time there's no null, and miss every time there's null. You'd end up with
hits=1, misses=1. You'd expect to get hits=1, misses=0, though. One way around
this specific example would be to know that "a","b" is due to a late arrival.
This would let you decrement the misses column. A more sophisticated example
would be hits, misses_a, and misses_b. In such a case, not only would you have
to know that "a", "b" was due to a late arrival, but you'd have to actually
know the last message as well ("a", NULL) -> ("a", "b"). I wonder if this is
something Millwheel provides that just wasn't talked about. I wonder if you
construct even more complex use cases where this model fully doesn't provide
exactly-once semantics.
I like that they simply said, "strong consistency," not, "exactly once." That
gives you a bit of wiggle room in these scenarios.
bq. 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.
Ya, we'd been discussing this internally. It's a little mind-bending for end
users, I think. Doesn't mean we can't do it. Just means we'd have to provide a
lot of good clear description about it.
bq. 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.
Agree. Another interesting thought is if we could integrate in some way with
audit, to allow us to know "for sure"-ish when a window is closed.
bq. Likewise you can choose whether to accept or reject one of these
updates--that is the user of the data gets to choose when the "close of books"
occurs, which seems better to me.
Yea. The way we'd been talking about windowing, I think this will work. We'd
been thinking of late arrivals (after punctuation) resulting in a drop. But if
the state is still persisted, there's nothing that says we couldn't emit the
"updated" late arrival value. Should be flexible.
One of the main advantages that Millwheel has is that it can store 2 weeks
worth of state. It's using a remote store (BigTable). Since we're local, this
might be a bit large. For aggregation, it's probably do-able, but for large
join jobs, 2 weeks worth of state might be many TBs of space. 100k msgs/sec@200
bytes/msg * 14 days = ~24 TB. Not un-doable, but would require a fair number of
machines to do this. If the 2 week state has a changelog attached, then it's
both local and in Kafka (with replication).
> 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)
