Interesting discussion. I think it is very important information, that when user will use a stateful ParDo, he can run into the situation where it will not behave correctly in "batch operating mode". But some transforms known to Beam, like fixed-window, will work fine? Is there a sorting applied to keyed elements before evaluating window key group? If answer is yes, then why not also do the same in case of stateful ParDo? It would feel consistent to me.
Part of SDK or not, I see DataFlow runner is doing this optimisation, probably precisely for making stateful ParDo operations stable in batch mode https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/BatchStatefulParDoOverrides.java#L64 On Thu, May 16, 2019 at 5:09 PM Jan Lukavský <je...@seznam.cz> wrote: > Hi Max, > answers inline. > ---------- Původní e-mail ---------- > Od: Maximilian Michels <m...@apache.org> > Komu: dev@beam.apache.org > Datum: 16. 5. 2019 15:59:59 > Předmět: Re: Definition of Unified model (WAS: Semantics of > PCollection.isBounded) > > Hi Jan, > > Thanks for the discussion. Aljoscha already gave great answers. Just a > couple of remarks: > > > a) streaming semantics (i.e. what I can express using Transforms) are > subset of batch semantics > > I think you mean streaming is a superset of batch, or batch is a subset > of streaming. This is the ideal. In practice, the two execution modes > are sometimes accomplished by two different execution engines. Even in > Flink, we have independent APIs for batch and streaming and the > execution semantics are slightly different. For example, there are no > watermarks in the batch API. Thus, batch rarely is simply an execution > mode of streaming. However, I still think the unified Beam model works > in both cases. > > > batch semantics and streaming semantics differs only in that I can have > GlobalWindow with default trigger on batch and cannot on stream > > Actually I really thought, that regarding semantics, streaming should be > subset of batch. That is because in batch, you can be sure that the > watermark will eventually approach infinity. That gives you one additional > feature, that streaming generally doesn't have (if you don't manually > forward watermark to infinity as you suggest). > > > > You can have a GlobalWindow in streaming with a default trigger. You > could define additional triggers that do early firings. And you could > even trigger the global window by advancing the watermark to +inf. > > Yes, but then you actually changed streaming to batch, you just execute > batch pipeline in streaming way. > > > > > On batch engines, this is generally not an issue, because the buffering > is eliminated by sorting - when a Group by operation occurs, batch runners > sort elements with the same key to be together and therefore eliminate the > need for potentially infinite cache. > > The batch engines you normally use might do that. However, I do not see > how sorting is an inherent property of the streaming model. We do not > guarantee a deterministic order of events in streaming with respect to > event time. In that regard, batch is a true subset of streaming because > we make no guarantees on the order. Actually, because we only advance > the watermark from -inf to +inf once we have read all data, this nicely > aligns with the streaming model. > > > Sure, streaming, doesn't have the time ordering guarantees. Having so > would be impractical. But - there is no issues in having these quarantees > in batch mode, moreover, it gives the pipelines, that need to have "bounded > out of orderness" the chance to ever finish. > > > I think that there is some issues in how we think about the properties of > batch vs. stream. If we define streaming as the superset, then we cannot > define some properties for batch, that streaming doesn't have. But - if we > just split it on the part of semantics and on the part of runtime > properties and guarantees, than it is possible to define properties of > batch, that streaming doesn't have. > > > Jan > > > > > -Max > > On 16.05.19 15:20, Aljoscha Krettek wrote: > > Hi, > > > > I think it’s helpful to consider that events never truly arrive in order > in the real world (you mentioned as much yourself). For streaming use > cases, there might be some out-of-orderness (or a lot of it, depending on > the use case) so your implementation has to be able to deal with that. On > the other end of the spectrum we have batch use cases, where > out-of-orderness is potentially even bigger because it allows for more > efficient parallel execution. If your implementation can deal with > out-of-orderness that also shouldn’t be a problem. > > > > Another angle is completeness vs. latency: you usually cannot have both > in a streaming world. If you want 100 % completeness, i.e. you want to > ensure that you process all events and never drop anything, you can never > advance the watermark from its initial -Inf if you want to also never have > watermark violations. In typical use cases I would expect any sorting > guarantees to be constantly violated, unless you are willing to drop late > data. > > > > I think these are some reasons why there is no mention of ordering by > timestamp anywhere (unless I’m mistaken and there is somewhere). > > > > You are right, of course, that batch-style runners can use > grouping/sorting for a GroupByKey operation. Flink does that and even > allows sorting by secondary key, so you could manually sort by timestamp as > a secondary key with hardly any additional cost. However, exposing that in > the model would make implementing Runners quite hard, or they would be > prohibitively slow. > > > > You’re also right that user functions that do arbitrary stateful > operations can be quite dangerous and lead to unexpected behaviour. You > example of reacting to changes in 0 and 1 would produce wrong results if > events are not 100% sorted by timestamp. In general, state changes that > rely on processing order are problematic while operations that move > monotonously though some space are fine. Examples of such operations are > adding elements to a set or summing numbers. If you “see” a given set of > events you can apply them to state in any order and as long as you see the > same set of events on different executions the result will be the same. > > > > As for the Beam execution model in relation to processing and time, I > think the only “guarantees” are: > > - you will eventually see all events > > - the timestamp of those events is usually not less than the watermark > (but not always) > > - the watermark will advance when the system thinks you won’t see events > with a smaller timestamp in the future (but you sometimes might) > > > > Those seem quite “poor”, but I think you can’t get better guarantees for > general cases for the reasons mentioned above. Also, this is just of the > top of my head and I might be wrong in my understanding of the Beam model. > :-O > > > > Best, > > Aljoscha > > > >> On 16. May 2019, at 13:53, Jan Lukavský <je...@seznam.cz> wrote: > >> > >> Hi, > >> > >> this is starting to be really exciting. It seems to me that there is > either something wrong with my definition of "Unified model" or with how it > is implemented inside (at least) Direct and Flink Runners. > >> > >> So, first what I see as properties of Unified model: > >> > >> a) streaming semantics (i.e. what I can express using Transforms) are > subset of batch semantics > >> > >> - this is true, batch semantics and streaming semantics differs only in > that I can have GlobalWindow with default trigger on batch and cannot on > stream > >> > >> b) runtime conditions of batch have to be subset of streaming > conditions > >> > >> - this is because otherwise it might be intractable to run streaming > pipeline on batch engine > >> > >> - generally this is also true - in batch mode watermark advances only > between two states (-inf and +inf), which makes it possible to turn (most) > stateful operations into group by key operations, and take advantage of > many other optimizations (ability to re-read inputs make it possible to > drop checkpointing, etc, etc) > >> > >> Now there is also one not so obvious runtime condition of streaming > engines - that is how skewed watermark and event time of elements being > processed can be - if this gets too high (i.e. watermark is not moving, > and/or elements are very out-of-order, then the processing might become > intractable, because everything might have to be buffered). > >> > >> On batch engines, this is generally not an issue, because the buffering > is eliminated by sorting - when a Group by operation occurs, batch runners > sort elements with the same key to be together and therefore eliminate the > need for potentially infinite cache. > >> > >> When this turns out to be an issue, is whenever there is a stateful > ParDo operation, because then (without sorting) there is violation of > property b) - on streaming engine the difference between element timestamp > and watermark will tend to be generally low (and late events will be > dropped to restrict the size of buffers), but on batch it can be > arbitrarily large and therefore size buffers that would be needed is > potentially unbounded. > >> > >> This line of thinking leads me to a conclusion, that if Beam doesn't > (on purpose) sort elements before stateful ParDo by timestamp, then it > basically violates the Unified model, because pipelines with stateful ParDo > will not function properly on batch engines. Which is what I observe - > there is non determinism on batch pipeline although everything seems to be > "well defined", elements arrive arbitrarily out of order and are > arbitrarily out of order dropped. This leads to different results everytime > batch pipeline is run. > >> > >> Looking forward to any comments on this. > >> > >> Jan > >> > >> On 5/16/19 10:53 AM, Aljoscha Krettek wrote: > >>> Please take this with a grain of salt, because I might be a bit rusty > on this. > >>> > >>> I think the Beam model does not prescribe any ordering (by time or > otherwise) on inputs. Mostly because always requiring it would be > prohibitively expensive on most Runners, especially global sorting. > >>> > >>> If you want to have sorting by key, you could do a GroupByKey and then > sort the groups in memory. This only works, of course, if your groups are > not too large. > >>> > >>>> On 15. May 2019, at 21:01, Jan Lukavský <je...@seznam.cz> wrote: > >>>> > >>>> Hmmm, looking into the code of FlinkRunner (and also by observing > results from the stateful ParDo), it seems, that I got it wrong from the > beginning. The data is not sorted before the stateful ParDo, but that a > little surprises me. How the operator should work in this case? It would > mean, that in the batch case I have to hold arbitrarily long > allowedLateness inside the BagState, which seems to be kind of suboptimal. > Or am I missing something obvious here? I'll describe the use case in more > detail, let's suppose I have a series of ones and zeros and I want emit at > each time point value of 1 if value changes from 0 to 1, value of -1 if > changes from 1 to 0 and 0 otherwise. So: > >>>> > >>>> 0, 1, 1, 0, 0, 1 -> 0, 1, 0, -1, 0, 1 > >>>> > >>>> Does anyone have a better idea how to solve it? And if not, how to > make it running on batch, without possibly infinite buffer? Should the > input to stateful ParDo be sorted in batch case? My intuition would be that > it should be, because in my understanding of "batch as a special case of > streaming" in batch case, there is (by default) single window, time > advances from -inf to +inf at the end, and the data contains no out of > order data, in places where this might matter (which therefore enables some > optimizations). The order would be relevant only in the stateful ParDo, I'd > say. > >>>> > >>>> Jan > >>>> > >>>> On 5/15/19 8:34 PM, Jan Lukavský wrote: > >>>>> Just to clarify, I understand, that changing semantics of the > PCollection.isBounded, is probably impossible now, because would probably > introduce chicken egg problem. Maybe I will state it more clearly - would > it be better to be able to run bounded pipelines using batch semantics on > DirectRunner (including sorting before stateful ParDos), or would it be > better to come up with some way to notify the pipeline that it will be > running in a streaming way although it consists only of bounded inputs? And > I'm not saying how to do it, just trying to find out if anyone else ever > had such a need. > >>>>> > >>>>> Jan > >>>>> > >>>>> On 5/15/19 5:20 PM, Jan Lukavský wrote: > >>>>>> Hi, > >>>>>> > >>>>>> I have come across unexpected (at least for me) behavior of some > apparent inconsistency of how a PCollection is processed in DirectRunner > and what PCollection.isBounded signals. Let me explain: > >>>>>> > >>>>>> - I have a stateful ParDo, which needs to make sure that elements > arrive in order - it accomplishes this by defining BagState for buffering > input elements and sorting them inside this buffer, it also keeps track of > element with highest timestamp to somehow estimate local watermark (minus > some allowed lateness), to know when to remove elements from the buffer, > sort them by time and pass them to some (time ordered) processing > >>>>>> > >>>>>> - this seems to work well for streaming (unbounded) data > >>>>>> > >>>>>> - for batch (bounded) data the semantics of stateful ParDo should > be (please correct me if I'm wrong) that elements always arrive in order, > because the runner can sort them by timestamp > >>>>>> > >>>>>> - this implies that for batch processed input (bounded) the > allowedLateness can be set to zero, so that the processing is little more > effective, because it doesn't have to use the BagState at all > >>>>>> > >>>>>> - now, the trouble seems to be, that DirectRunner always uses > streaming processing, even if the input is bounded (that is by definition > possible), but there is no way now to know when it is possible to change > allowed lateness to zero (because input will arrive ordered) > >>>>>> > >>>>>> - so - it seems to me, that either DirectRunner should apply > sorting to stateful ParDo, when it processes bounded data (the same way > that other runners do), or it can apply streaming processing, but then it > should change PCollection.isBounded to UNBOUNDED, even if the input is > originally bounded > >>>>>> > >>>>>> - that way, the semantics of PCollection.isBounded, would be not if > the data are known in advance to be finite, but *how* the data are going to > be processed, which is much more valuable (IMO) > >>>>>> > >>>>>> Any thoughts? > >>>>>> > >>>>>> Jan > >>>>>> > > > >
