On Wed, Feb 17, 2021 at 1:56 PM Kenneth Knowles <k...@apache.org
<mailto:k...@apache.org>> wrote:
On Wed, Feb 17, 2021 at 1:06 PM Robert Bradshaw
<rober...@google.com <mailto:rober...@google.com>> wrote:
I would prefer to leave downstream triggering up to the
runner (or, better, leave upstream triggering up to the
runner, a la sink triggers), but one problem is that without
an explicit AfterSynchronizedProcessingTime one can't tell if
the downstream ProcessingTime between two groupings is due to
an explicit re-triggering between them or inherited from one
to the other.
I mean to propose that there should be no triggering specified
unless due to explicit re-triggering.
You're saying that we leave the trigger (and perhaps other) fields of
the WindowingStrategy attached to PCollections downstream the first
GBK unset in the proto? And let runners walk over the graph to infer
it? I could be OK with making this legal, though updating all SDKs
and Runners to handle this doesn't seem high priority at the moment.
(and BTW yes I agree about sink triggers, but we need retractions
and probably some theoretical work before we can aim for that)
Kenn
On Wed, Feb 17, 2021 at 12:37 PM Kenneth Knowles
<k...@apache.org <mailto:k...@apache.org>> wrote:
Just for the thread I want to comment on another, more
drastic approach: eliminate continuation triggers from
the model, leaving downstream triggering up to a runner.
This approach is not viable because transforms may need
to change their behavior based on whether or not a
trigger will fire more than once. Transforms can and do
inspect the windowing strategy to do things differently.
Kenn
On Wed, Feb 17, 2021 at 11:47 AM Reuven Lax
<re...@google.com <mailto:re...@google.com>> wrote:
I'll say that synchronized processing time has
confused users before. Users sometimes use
processing-time triggers to optimize latency, banking
that that will decouple stage latency from the
long-tail latency of previous stages. However
continuation triggers silently switching to
synchronized processing time has defeated that, and
it wasn't clear to users why.
On Wed, Feb 17, 2021 at 11:12 AM Robert Bradshaw
<rober...@google.com <mailto:rober...@google.com>> wrote:
On Fri, Feb 12, 2021 at 9:09 AM Kenneth Knowles
<k...@apache.org <mailto:k...@apache.org>> wrote:
On Thu, Feb 11, 2021 at 9:38 PM Robert
Bradshaw <rober...@google.com
<mailto:rober...@google.com>> wrote:
Of course the right answer is to just
implement sink triggers and sidestep the
question altogether :).
In the meantime, I think leaving
AfterSynchronizedProcessingTime in the
model makes the most sense, and runners
can choose an implementation between
firing eagerly and waiting some amount of
time until they think all (most?)
downstream results are in before firing,
depending on how smart the runner wants
to be. As you point out, they're all
correct, and we'll have multiple firings
due to the upstream trigger anyway, and
this is safer than it used to be (though
still possibly requires work).
Just to clarify, as I got a little confused,
is your suggestion: Leave
AfterSynchronizedProcessingTime* triggers in
the model/proto, let the SDK put them in
where they want, and let runners decide how
to interpret them? (this SGTM and requires
the least/no changes)
Yep. We may want to update Python/Go to produce
AfterSynchronizedProcessingTime downstream of
ProcessingTime triggers too, eventually, to
better express intent.
Kenn
*noting that
TimeDomain.SYNCHRONIZED_PROCESSING_TIME is
not related to this, except in
implementation, and should be removed either way.
On Wed, Feb 10, 2021 at 1:37 PM Kenneth
Knowles <k...@apache.org
<mailto:k...@apache.org>> wrote:
Hi all,
TL;DR:
1. should we replace "after
synchronized processing time" with
"after count 1"?
2. should we remove "continuation
trigger" and leave this to runners?
----
"AfterSynchronizedProcessingTime"
triggers were invented to solve a
specific problem. They are
inconsistent across SDKs today.
- You have an aggregation/GBK with
aligned processing time trigger like
("output every minute on the minute")
- You have a downstream
aggregation/GBK between that and the sink
- You expect to have about one
output every minute per key+window pair
Any output of the upstream
aggregation may contribute to any
key+window of the downstream
aggregation. The
AfterSynchronizedProcessingTime
trigger waits for all the processing
time based triggers to fire and
commit their outputs. The downstream
aggregation will output as fast as
possible in panes consistent with the
upstream aggregation.
- The Java SDK behavior is as above,
to output "as fast as reasonable".
- The Python SDK never uses
"AfterSynchronizedProcessingTime"
triggers but simply propagates the
same trigger to the next GBK,
creating additional delay.
- I don't know what the Go SDK may
do, if it supports this at all.
Any behavior could be defined as
"correct". A simple option could be
to have the downstream aggregation
"fire always" aka "after element
count 1". How would this change
things? We would potentially see many
more outputs.
Why did we do this in the first
place? There are (at least) these
reasons:
- Previously, triggers could
"finish" an aggregation thus dropping
all further data. In this case,
waiting for all outputs is critical
or else you lose data. Now triggers
cannot finish aggregations.
- Whenever there may be more than
one pane, a user has to write logic
to compensate and deal with it.
Changing from guaranteed single pane
to multi-pane would break things. So
if the user configures a single
firing, all downstream aggregations
must respect it. Now that triggers
cannot finish, I think processing
time can only be used in multi-pane
contexts anyhow.
- The above example illustrates how
the behavior in Java maintains
something that the user will expect.
Or so we think. Maybe users don't care.
How did we get into this inconsistent
state? When the user specifies
triggering it applies to the very
nearest aggregation/GBK. The SDK
decides what triggering to insert
downstream. One possibility is to
remove this and have it unspecified,
left to runner behavior.
I think maybe these pieces of
complexity are both not helpful and
also not (necessarily) breaking
changes to alter, especially
considering we have inconsistency in
the model.
WDYT? And I wonder what this means
for xlang and portability... how does
continuation triggering even work?
(if at all)
Kenn
