Yes - I don't think we should try and make any deterministic guarantees about what is in a bundle. Stability guarantees are per element only.
On Wed, Aug 9, 2017 at 1:30 PM, Thomas Groh <tg...@google.com.invalid> wrote: > +1 to the annotation-on-ProcessElement approach. ProcessElement is the > minimum implementation requirement of a DoFn, and should be where the > processing logic which depends on characteristics of the inputs lie. It's a > good way of signalling the requirements of the Fn, and letting the runner > decide. > > I have a minor concern that this may not work as expected for users that > try to batch remote calls in `FinishBundle` - we should make sure we > document that it is explicitly the input elements that will be replayed, > and bundles and other operational are still arbitrary. > > > > On Wed, Aug 9, 2017 at 10:37 AM, Reuven Lax <re...@google.com.invalid> > wrote: > > > I think deterministic here means deterministically replayable. i.e. no > > matter how many times the element is retried, it will always be the same. > > > > I think we should also allow specifying this on processTimer. This would > > mean that any keyed state written in a previous processElement must be > > guaranteed durable before processTimer is called. > > > > > > On Wed, Aug 9, 2017 at 10:10 AM, Ben Chambers <bchamb...@apache.org> > > wrote: > > > > > I strongly agree with this proposal. I think moving away from "just > > insert > > > a GroupByKey for one of the 3 different reasons you may want it" > towards > > > APIs that allow code to express the requirements they have and the > runner > > > to choose the best way to meet this is a major step forwards in terms > of > > > portability. > > > > > > I think "deterministic" may be misleading. The actual contents of the > > > collection aren't deterministic if upstream computations aren't. The > > > property we really need is that once an input may have been observed by > > the > > > side-effecting code it will never be observed with a different value. > > > > > > I would propose something RequiresStableInput, to indicate that the > input > > > must be stable as observed by the function. I could also see something > > > hinting at the fact we don't recompute the input, such as > > > RequiresMemoizedInput or RequiresNoRecomputation. > > > > > > -- Ben > > > > > > P.S For anyone interested other uses of GroupByKey that we may want to > > > discuss APIs for would be preventing retry across steps (eg., > preventing > > > fusion) and redistributing inputs across workers. > > > > > > On Wed, Aug 9, 2017 at 9:53 AM Kenneth Knowles <k...@google.com.invalid > > > > > wrote: > > > > > > > This came up again, so I wanted to push it along by proposing a > > specific > > > > API for Java that could have a derived API in Python. I am writing > this > > > > quickly to get something out there, so I welcome suggestions for > > > revision. > > > > > > > > Today a DoFn has a @ProcessElement annotated method with various > > > automated > > > > parameters, but most fundamentally this: > > > > > > > > @ProcessElement > > > > public void process(ProcessContext ctx) { > > > > ctx.element() // to access the current input element > > > > ctx.output(something) // to write to default output collection > > > > ctx.output(tag, something) // to write to other output collections > > > > } > > > > > > > > For some time, we have hoped to unpack the context - it is a > > > > backwards-compatibility pattern made obsolete by the new DoFn design. > > So > > > > instead it would look like this: > > > > > > > > @ProcessElement > > > > public void process(Element element, MainOutput mainOutput, ...) { > > > > element.get() // to access the current input element > > > > mainOutput.output(something) // to write to the default output > > > collection > > > > other.output(something) // to write to other output collection > > > > } > > > > > > > > I've deliberately left out the undecided syntax for side outputs. But > > it > > > > would be nice for the tag to be built in to the parameter so it > doesn't > > > > have to be used when calling output(). > > > > > > > > One way to enhance this to deterministic input would just be this: > > > > > > > > @ProcessElement > > > > @RequiresDeterministicInput > > > > public void process(Element element, MainOutput mainOutput, ...) { > > > > element.get() // to access the current input element > > > > mainOutput.output(something) // to write to the default output > > > collection > > > > other.output(something) // to write to other output collection > > > > } > > > > > > > > There are really a lot of places where we could put an annotation or > > > change > > > > a type to indicate that the input PCollection should be > > > > well-defined/deterministically-replayable. I don't have a really > > strong > > > > opinion. > > > > > > > > Kenn > > > > > > > > On Tue, Mar 21, 2017 at 4:53 PM, Ben Chambers > > > <bchamb...@google.com.invalid > > > > > > > > > wrote: > > > > > > > > > Allowing an annotation on DoFn's that produce deterministic output > > > could > > > > be > > > > > added in the future, but doesn't seem like a great option. > > > > > > > > > > 1. It is a correctness issue to assume a DoFn is deterministic and > be > > > > > wrong, so we would need to assume all transform outputs are > > > > > non-deterministic unless annotated. Getting this correct is > difficult > > > > (for > > > > > example, GBK is surprisingly non-deterministic except in specific > > > cases). > > > > > > > > > > 2. It is unlikely to be a major performance improvement, given that > > any > > > > > non-deterministic transform prior to a sink (which are most likely > to > > > > > require deterministic input) will cause additional work to be > needed. > > > > > > > > > > Based on this, it seems like the risk of allowing an annotation is > > high > > > > > while the potential for performance improvements is low. The > current > > > > > proposal (not allowing an annotation) makes sense for now, until we > > can > > > > > demonstrate that the impact on performance is high in cases that > > could > > > be > > > > > avoided with an annotation (in real-world use). > > > > > > > > > > -- Ben > > > > > > > > > > On Tue, Mar 21, 2017 at 2:05 PM vikas rk <vikky...@gmail.com> > wrote: > > > > > > > > > > +1 for the general idea of runners handling it over hard-coded > > > > > implementation strategy. > > > > > > > > > > For the Write transform I believe you are talking about > > > ApplyShardingKey > > > > > < > > > > > https://github.com/apache/beam/blob/d66029cafde152c0a46ebd276ddfa4 > > > > > c3e7fd3433/sdks/java/core/src/main/java/org/apache/beam/sdk/ > > > > > io/Write.java#L304 > > > > > > > > > > > which > > > > > introduces non deterministic behavior when retried? > > > > > > > > > > > > > > > *Let a DoFn declare (mechanism not important right now) that it > > > > > "requiresdeterministic input"* > > > > > > > > > > > > > > > > > > > > *Each runner will need a way to induce deterministic input - the > > > > > obviouschoice being a materialization.* > > > > > > > > > > Does this mean that a runner will always materialize (or whatever > the > > > > > strategy is) an input PCollection to this DoFn even though the > > > > PCollection > > > > > might have been produced by deterministic transforms? Would it make > > > sense > > > > > to also let DoFns declare if they produce non-deterministic output? > > > > > > > > > > -Vikas > > > > > > > > > > > > > > > On 21 March 2017 at 13:52, Stephen Sisk <s...@google.com.invalid> > > > wrote: > > > > > > > > > > > Hey Kenn- > > > > > > > > > > > > this seems important, but I don't have all the context on what > the > > > > > problem > > > > > > is. > > > > > > > > > > > > Can you explain this sentence "Specifically, there is > pseudorandom > > > data > > > > > > generated and once it has been observed and used to produce a > side > > > > > effect, > > > > > > it cannot be regenerated without erroneous results." ? > > > > > > > > > > > > Where is the pseudorandom data coming from? Perhaps a concrete > > > example > > > > > > would help? > > > > > > > > > > > > S > > > > > > > > > > > > > > > > > > On Tue, Mar 21, 2017 at 1:22 PM Kenneth Knowles > > > <k...@google.com.invalid > > > > > > > > > > > wrote: > > > > > > > > > > > > > Problem: > > > > > > > > > > > > > > I will drop all nuance and say that the `Write` transform as it > > > > exists > > > > > in > > > > > > > the SDK is incorrect until we add some specification and APIs. > We > > > > can't > > > > > > > keep shipping an SDK with an unsafe transform in it, and IMO > this > > > > > > certainly > > > > > > > blocks a stable release. > > > > > > > > > > > > > > Specifically, there is pseudorandom data generated and once it > > has > > > > been > > > > > > > observed and used to produce a side effect, it cannot be > > > regenerated > > > > > > > without erroneous results. > > > > > > > > > > > > > > This generalizes: For some side-effecting user-defined > functions, > > > it > > > > is > > > > > > > vital that even across retries/replays they have a consistent > > view > > > of > > > > > the > > > > > > > contents of their input PCollection, because their effect on > the > > > > > outside > > > > > > > world cannot be retracted if/when they fail and are retried. > Once > > > the > > > > > > > runner ensures a consistent view of the input, it is then their > > own > > > > > > > responsibility to be idempotent. > > > > > > > > > > > > > > Ideally we should specify this requirement for the user-defined > > > > > function > > > > > > > without imposing any particular implementation strategy on Beam > > > > > runners. > > > > > > > > > > > > > > Proposal: > > > > > > > > > > > > > > 1. Let a DoFn declare (mechanism not important right now) that > it > > > > > > "requires > > > > > > > deterministic input". > > > > > > > > > > > > > > 2. Each runner will need a way to induce deterministic input - > > the > > > > > > obvious > > > > > > > choice being a materialization. > > > > > > > > > > > > > > I want to keep the discussion focused, so I'm leaving out any > > > > > > possibilities > > > > > > > of taking this further. > > > > > > > > > > > > > > Regarding performance: Today places that require this tend to > be > > > > > already > > > > > > > paying the cost via GroupByKey / Reshuffle operations, since > that > > > > was a > > > > > > > simple way to induce determinism in batch Dataflow* (doesn't > work > > > for > > > > > > most > > > > > > > other runners nor for streaming Dataflow). This change will > > > replace a > > > > > > > hard-coded implementation strategy with a requirement that may > be > > > > > > fulfilled > > > > > > > in the most efficient way available. > > > > > > > > > > > > > > Thoughts? > > > > > > > > > > > > > > Kenn (w/ lots of consult from colleagues, especially Ben) > > > > > > > > > > > > > > * There is some overlap with the reshuffle/redistribute > > discussion > > > > > > because > > > > > > > of this historical situation, but I would like to leave that > > > broader > > > > > > > discussion out of this correctness issue. > > > > > > > > > > > > > > > > > > > > > > > > > > > >