On Mon, May 20, 2019 at 1:19 PM Jan Lukavský <je...@seznam.cz> wrote: > > Hi Robert, > > yes, I think you rephrased my point - although no *explicit* guarantees > of ordering are given in either mode, there is *implicit* ordering in > streaming case that is due to nature of the processing - the difference > between watermark and timestamp of elements flowing through the pipeline > are generally low (too high difference leads to the overbuffering > problem), but there is no such bound on batch.
Fortunately, for batch, only the state for a single key needs to be preserved at a time, rather than the state for all keys across the range of skew. Of course if you have few or hot keys, one can still have issues (and this is not specific to StatefulDoFns). > As a result, I see a few possible solutions: > > - the best and most natural seems to be extension of the model, so > that it defines batch as not only "streaming pipeline executed in batch > fashion", but "pipeline with at least as good runtime characteristics as > in streaming case, executed in batch fashion", I really don't think that > there are any conflicts with the current model, or that this could > affect performance, because the required sorting (as pointed by > Aljoscha) is very probably already done during translation of stateful > pardos. Also note that this definition only affects user defined > stateful pardos I don't see batch vs. streaming as part of the model. One can have microbatch, or even a runner that alternates between different modes. The model describes what the valid outputs are given a (sometimes partial) set of inputs. It becomes really hard to define things like "as good runtime characteristics." Once you allow any out-of-orderedness, it is not very feasible to try and define (and more cheaply implement) a "upper bound" of acceptable out-of-orderedness. Pipelines that fail in the "worst case" batch scenario are likely to degrade poorly (possibly catastrophically) when the watermark falls behind in streaming mode as well. > - another option would be to introduce annotation for DoFns (e.g. > @RequiresStableTimeCharacteristics), which would result in the sorting > in batch case - but - this extension would have to ensure the sorting in > streaming mode also - it would require definition of allowed lateness, > and triggger (essentially similar to window) This might be reasonable, implemented by default by buffering everything and releasing elements as the watermark (+lateness) advances, but would likely lead to inefficient (though *maybe* easier to reason about) code. Not sure about the semantics of triggering here, especially data-driven triggers. Would it be roughly equivalent to GBK + FlatMap(lambda (key, values): [(key, value) for value in values])? Or is the underlying desire just to be able to hint to the runner that the code may perform better (e.g. require less resources) as skew is reduced (and hence to order by timestamp iff it's cheap)? > - last option would be to introduce these "higher order guarantees" in > some extension DSL (e.g. Euphoria), but that seems to be the worst > option to me > > I see the first two options quite equally good, although the letter one > is probably more time consuming to implement. But it would bring > additional feature to streaming case as well. > > Thanks for any thoughts. > > Jan > > On 5/20/19 12:41 PM, Robert Bradshaw wrote: > > On Fri, May 17, 2019 at 4:48 PM Jan Lukavský <je...@seznam.cz> wrote: > >> Hi Reuven, > >> > >>> How so? AFAIK stateful DoFns work just fine in batch runners. > >> Stateful ParDo works in batch as far, as the logic inside the state works > >> for absolutely unbounded out-of-orderness of elements. That basically > >> (practically) can work only for cases, where the order of input elements > >> doesn't matter. But, "state" can refer to "state machine", and any time > >> you have a state machine involved, then the ordering of elements would > >> matter. > > No guarantees on order are provided in *either* streaming or batch > > mode by the model. However, it is the case that in order to make > > forward progress most streaming runners attempt to limit the amount of > > out-of-orderedness of elements (in terms of event time vs. processing > > time) to make forward progress, which in turn could help cap the > > amount of state that must be held concurrently, whereas a batch runner > > may not allow any state to be safely discarded until the whole > > timeline from infinite past to infinite future has been observed. > > > > Also, as pointed out, state is not preserved "batch to batch" in batch mode. > > > > > > On Thu, May 16, 2019 at 3:59 PM Maximilian Michels <m...@apache.org> wrote: > > > >>> batch semantics and streaming semantics differs only in that I can have > >>> GlobalWindow with default trigger on batch and cannot on stream > >> 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. > > IIRC, as a pragmatic note, we prohibited global window with default > > trigger on unbounded PCollections in the SDK because this is more > > likely to be user error than an actual desire to have no output until > > drain. But it's semantically valid in the model.
