On Tue, Nov 26, 2019 at 1:00 AM Jan Lukavský <[email protected]
<mailto:[email protected]>> wrote:
> I will not try to formalize this notion in this email. But I
will note that since it is universally assured, it would be zero
cost and significantly safer to formalize it and add an annotation
noting it was required. It has nothing to do with event time
ordering, only trigger firing ordering.
I cannot agree with the last sentence (and I'm really not doing
this on purpose :-)). Panes generally arrive out of order, as
mentioned several times in the discussions linked from this
thread. If we want to ensure "trigger firing ordering", we can use
the pane index, that is correct. But - that is actually equivalent
to sorting by event time, because pane index order will be
(nearly) the same as event time order. This is due to the fact,
that pane index and event time correlate (both are monotonic).
Trigger firings can have decreasing event timestamps w/ the minimum
timestamp combiner*. I do think the issue at hand is best analyzed in
terms of the explicit ordering on panes. And I do think we need to
have an explicit guarantee or annotation strong enough to describe a
correct-under-all-allowed runners sink. Today an antagonistic runner
could probably break a lot of things.
Kenn
*In fact, they can decrease via the "maximum" timestamp combiner
because actually timestamp combiners only apply to the elements that
particular pane. This is weird, and maybe a design bug, but good to
know about.
The pane index "only" solves the issue of preserving ordering even
in case where there are multiple firings within the same timestamp
(regardless of granularity). This was mentioned in the initial
discussion about event time ordering, and is part of the design
doc - users should be allowed to provide UDF for extracting
time-correlated ordering field (which means ability to choose a
preferred, or authoritative, observer which assigns unambiguous
ordering to events). Example of this might include Kafka offsets
as well, or any queue index for that matter. This is not yet
implemented, but could (should) be in the future.
The only case where these two things are (somewhat) different is
the case mentioned by @Steve - if the output is stateless ParDo,
which will get fused. But that is only because the processing is
single-threaded per key, and therefore the ordering is implied by
timer ordering (and careful here, many runners don't have this
ordering 100% correct, as of now - this problem luckily appears
only when there are multiple timers per key). Moreover, if there
should be a failure, then the output might (would) get back in
time anyway. If there would be a shuffle operation after
GBK/Combine, then the ordering is no longer guaranteed and must be
explicitly taken care of.
Last note, I must agree with @Rui that all these discussions are
very much related to retractions (precisely the ability to
implement them).
Jan
On 11/26/19 7:34 AM, Kenneth Knowles wrote:
Hi Aaron,
Another insightful observation.
Whenever an aggregation (GBK / Combine per key) has a trigger
firing, there is a per-key sequence number attached. It is
included in metadata known as "PaneInfo" [1]. The value of
PaneInfo.getIndex() is colloquially referred to as the "pane
index". You can also make use of the "on time index" if you like.
The best way to access this metadata is to add a parameter of
type PaneInfo to your DoFn's @ProcessElement method. This works
for stateful or stateless DoFn.
Most of Beam's IO connectors do not explicitly enforce that
outputs occur in pane index order but instead rely on the hope
that the runner delivers panes in order to the sink. IMO this is
dangerous but it has not yet caused a known issue. In practice,
each "input key to output key 'path' " through a pipeline's logic
does preserve order for all existing runners AFAIK and it is the
formalization that is missing. It is related to an observation by
+Rui Wang <mailto:[email protected]> that processing retractions
requires the same key-to-key ordering.
I will not try to formalize this notion in this email. But I will
note that since it is universally assured, it would be zero cost
and significantly safer to formalize it and add an annotation
noting it was required. It has nothing to do with event time
ordering, only trigger firing ordering.
Kenn
[1]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/PaneInfo.java
[2]
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java#L557
On Mon, Nov 25, 2019 at 4:06 PM Pablo Estrada <[email protected]
<mailto:[email protected]>> wrote:
The blog posts on stateful and timely computation with Beam
should help clarify a lot about how to use state and timers
to do this:
https://beam.apache.org/blog/2017/02/13/stateful-processing.html
https://beam.apache.org/blog/2017/08/28/timely-processing.html
You'll see there how there's an implicit per-single-element
grouping for each key, so state and timers should support
your use case very well.
Best
-P.
On Mon, Nov 25, 2019 at 3:47 PM Steve Niemitz
<[email protected] <mailto:[email protected]>> wrote:
If you have a pipeline that looks like Input ->
GroupByKey -> ParDo, while it is not guaranteed, in
practice the sink will observe the trigger firings in
order (per key), since it'll be fused to the output of
the GBK operation (in all runners I know of).
There have been a couple threads about trigger ordering
as well on the list recently that might have more
information:
https://lists.apache.org/thread.html/b61a908289a692dbd80dd6a869759eacd45b308cb3873bfb77c4def6@%3Cdev.beam.apache.org%3E
https://lists.apache.org/thread.html/20d11046d26174969ef44a781e409a1cb9f7c736e605fa40fdf98397@%3Cuser.beam.apache.org%3E
On Mon, Nov 25, 2019 at 5:52 PM Aaron Dixon
<[email protected] <mailto:[email protected]>> wrote:
@Jan @Pablo Thank you
@Pablo In this case it's a single global windowed
Combine/perKey, triggered per element. Keys are few
(client accounts) so they can live forever.
It looks like just by virtue of using a stateful
ParDo I could get this final execution to be
"serialized" per key. (Then I could simply do the
compare-and-swap using Beam's state mechanism to keep
track of the "latest trigger timestamp" instead of
having to orchestrate compare-and-swap in the target
store :thinking:.)
On Mon, Nov 25, 2019 at 4:14 PM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
One addition, to make the list of options
exhaustive, there is probably
one more option
c) create a ParDo keyed by primary key of your
sink, cache the last
write in there and compare it locally, without
the need to query the
database
It would still need some timer to clear values
after watermark + allowed
lateness, because otherwise you would have to
cache your whole database
on workers. But because you don't need actual
ordering, you just need
the most recent value (if I got it right) this
might be an option.
Jan
On 11/25/19 10:53 PM, Jan Lukavský wrote:
> Hi Aaron,
>
> maybe someone else will give another option,
but if I understand
> correctly what you want to solve, then you
essentially have to do either:
>
> a) use the compare & swap mechanism in the
sink you described
>
> b) use a buffer to buffer elements inside the
outputting ParDo and
> only output them when watermark passes (using a
timer).
>
> There is actually an ongoing discussion about
how to make option b)
> user-friendly and part of Beam itself, but
currently there is no
> out-of-the-box solution for that.
>
> Jan
>
> On 11/25/19 10:27 PM, Aaron Dixon wrote:
>> Suppose I trigger a Combine per-element (in a
high-volume stream) and
>> use a ParDo as a sink.
>>
>> I assume there is no guarantee about the order
that my ParDo will see
>> these triggers, especially as it processes in
parallel, anyway.
>>
>> That said, my sink writes to a db or cache and
I would not like the
>> cache to ever regress its value to something
"before" what it has
>> already written.
>>
>> Is the best way to solve this problem to
always write the event-time
>> in the cache and do a compare-and-swap only
updating the sink if the
>> triggered value in-hand is later than the
target value?
>>
>> Or is there a better way to guarantee that my
ParDo sink will process
>> elements in-order? (Eg, if I can give up
per-event/real-time, then a
>> delay-based trigger would probably be
sufficient I imagine.)
>>
>> Thanks for advice!
