On Sat, Oct 22, 2016 at 2:38 AM, Amit Sela <amitsel...@gmail.com> wrote: > I understand the semantics, but I feel like there might be a different > point of view for open-source runners.
It seems we're losing a major promise of the runner interchangeability story if different runners can give different results for a well-defined transformation. I strongly feel we should avoid that path whenever possible. Specifically in this case Combine.perKey should mean the same thing on all runners (namely its composite definition), and only be executed differently when it's safe to do so. > Dataflow is a service, and it tries to do it's best to optimize execution > while users don't have to worry about internal implementation (they are not > aware of it). > I can assure > <https://databricks.gitbooks.io/databricks-spark-knowledge-base/content/best_practices/prefer_reducebykey_over_groupbykey.html> > you that for Spark users, applying groupByKey instead of combinePerKey is > an important note. For sure. Dataflow calls this out too. See the second star at https://cloud.google.com/dataflow/model/combine#using-combine-perkey (though it's not called out as prominently as it is for Spark users--likely should be more). Beam documentation should make this point as well. > @Aljoscha do Flink users (working on Flink native API) usually care about > this difference of implementation ? > Any other runners that can provide input ? IIRC, Flink and Dataflow (and, trivially, the direct runner) all avoid this unsafe optimization when merging windows are mixed with non-global side inputs. Note also that the user of the Combine.perKey transform may not know the choice of windowing of the main or side inputs, so can't make this determination of whether it's safe to use this optimization. (As a concrete example, suppose I created a TopNPercent transform that did a global count and passed that as a side input to the Top CombineFn.) > On Sat, Oct 22, 2016 at 2:25 AM Robert Bradshaw <rober...@google.com.invalid> > wrote: > > Combine.perKey() is defined as GroupByKey() | Combine.values(). > > A runner is free, in fact encouraged, to take advantage of the > associative properties of CombineFn to compute the result of > GroupByKey() | Combine.values() as cheaply as possible, but it is > incorrect to produce something that could not have been produced by > this composite implementation. (In the case of deterministic trigger > firing, (e.g. the default trigger), plus assuming of course a > associative, deterministic CombineFn, there is exactly one correct > output for every input no matter the WindowFns). > > A corollary to this is that we cannot apply combining operations that > inspect the main input window (including side inputs where the mapping > is anything but the constant map (like to GlobalWindow)) until the > main input window is known. > > > On Fri, Oct 21, 2016 at 3:50 PM, Amit Sela <amitsel...@gmail.com> wrote: >> Please excuse my typos and apply "s/differ/defer/g" ;-). >> Amit. >> >> On Fri, Oct 21, 2016 at 2:59 PM Amit Sela <amitsel...@gmail.com> wrote: >> >>> I'd like to raise an issue that was discussed in BEAM-696 >>> <https://issues.apache.org/jira/browse/BEAM-696>. >>> I won't recap here because it would be extensive (and probably >>> exhaustive), and I'd also like to restart the discussion here rather then >>> summarize it. >>> >>> *The problem* >>> In the case of (main) input in a merging window (e.g. Sessions) with >>> sideInputs, pre-combining might lead to non-deterministic behaviour, for >>> example: >>> Main input: e1 (time: 3), e2 (time: 5) >>> Session: gap duration of 3 -> e1 alone belongs to [3, 6), e2 alone [5, > 8), >>> combined together the merging of their windows yields [3, 8). >>> Matching SideInputs with FixedWindows of size 2 should yield - e1 > matching >>> sideInput window [4, 6), e2 [6, 8), merged [6, 8). >>> Now, if the sideInput is used in a merging step of the combine, and both >>> elements are a part of the same bundle, the sideInput accessed will >>> correspond to [6, 8) which is the expected behaviour, but if e1 is >>> pre-combined in a separate bundle, it will access sideInput for [4, 6) >>> which is wrong. >>> ** this can tends to be a bit confusing, so any > clarifications/corrections >>> are most welcomed.* >>> >>> *Solutions* >>> The optimal solution would be to differ until trigger in case of merging >>> windows with sideInputs that are not "agnostic" to such behaviour, but > this >>> is clearly not feasible since the nature and use of sideInputs in >>> CombineFns are opaque. >>> Second best would be to differ until trigger *only* if sideInputs are >>> used for merging windows - pretty sure this is how Flink and Dataflow > (soon >>> Spark) runners do that. >>> >>> *Tradeoffs* >>> This seems like a very user-friendly way to apply authored pipelines >>> correctly, but this also means that users who called for a Combine >>> transformation will get a Grouping transformation instead (sort of the >>> opposite of combiner lifting ? a combiner unwrapping ?). >>> For the SDK, Combine is simply a composite transform, but keep in mind >>> that this affects runner optimization. >>> The price to pay here is (1) shuffle all elements into a single bundle >>> (the cost varies according to a runner's typical bundle size) (2) state > can >>> grow as processing is differed and not compacted until triggered. >>> >>> IMHO, the execution should remain faithful to what the pipeline states, >>> and if this results in errors, well... it happens. >>> There are many legitimate use cases where an actual GroupByKey should be >>> used (regardless of sideInputs), such as sequencing of events in a > window, >>> and I don't see the difference here. >>> >>> As stated above, I'm (almost) not recapping anyones notes as they are >>> persisted in BEAM-696, so if you had something to say please provide you >>> input here. >>> I will note that Ben Chambers and Pei He mentioned that even with >>> differing, this could still run into some non-determinism if there are >>> triggers controlling when we extract output because non-merging windows' >>> trigger firing is non-deterministic. >>> >>> Thanks, >>> Amit >>> >>>
