so I think the following will do exactly that and can be easily factored into a reusable transform (modulo Java type boilerplate):
pCollection.apply(WithKeys.of((Element e) -> ThreadLocalRandom.current().nextInt(N)) .apply(Window.into(new GlobalWindows()) .triggering(AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(100)))) .apply(GroupByKey.create()) .apply(ParDo.of(new DoFn<>() { @ProcessElement @StableReplay public void processElement(ProcessContext c) { // Insert c.element().getValue() into backend. } }); On Wed, Nov 15, 2017 at 7:00 PM, Romain Manni-Bucau <rmannibu...@gmail.com> wrote: > 2017-11-15 11:42 GMT+01:00 Reuven Lax <re...@google.com.invalid>: > > Can we describe this at a higher level? > > > > I think what you want is the following. Please correct if I'm > > misunderstanding. > > > > Batches of 100 elements (is this a hard requirement, or do they have to > be > > "approximately" 100 element?) > > Approximately is fine while documented (what is not is 1000000 instead > of 10 for instance) > > > > > Once you see a batch, you're guaranteed to see the same batch on retries. > > +1 > > > > > You want to then idempotently insert this batch into some backend. Things > > may fail, workers may crash, but in that case you want to get the exact > > same batch back so you can insert it again. > > +1 > > > > > Do you care about ordering? On failure do you have to see the same > batches > > in the same order as before, or is it sufficient to see the same batches? > > Beam doesnt everywhere so I guess it is not important - at least for > my cases this statement is true. > > > > > Reuven > > > > On Wed, Nov 15, 2017 at 5:58 PM, Romain Manni-Bucau < > rmannibu...@gmail.com> > > wrote: > > > >> Overall goal is to ensure each 100 elements max, a "backend" (as > >> datastore) flush/commit/push is done and is aligned with beam > >> checkpoints. You can see it as bringing the "general" commit-interval > >> notion to beam and kind of get rid of the bundle notion which is > >> almost impossible to use today. > >> > >> Romain Manni-Bucau > >> @rmannibucau | Blog | Old Blog | Github | LinkedIn > >> > >> > >> 2017-11-15 10:27 GMT+01:00 Reuven Lax <re...@google.com.invalid>: > >> > It's in the dev list archives, not sure if there's a doc yet. > >> > > >> > I'm not quite sure I understand what you mean by a "flush" Can you > >> describe > >> > the problem you're trying to solve? > >> > > >> > Reuven > >> > > >> > On Wed, Nov 15, 2017 at 5:25 PM, Romain Manni-Bucau < > >> rmannibu...@gmail.com> > >> > wrote: > >> > > >> >> Hmm, I didn't find the doc - if you have the link not far it would be > >> >> appreciated - but "before" sounds not enough, it should be "after" in > >> >> case there was a "flush" no? > >> >> > >> >> Romain Manni-Bucau > >> >> @rmannibucau | Blog | Old Blog | Github | LinkedIn > >> >> > >> >> > >> >> 2017-11-15 10:10 GMT+01:00 Reuven Lax <re...@google.com.invalid>: > >> >> > If you set @StableReplay before a ParDo, it forces a checkpoint > before > >> >> that > >> >> > ParDo. > >> >> > > >> >> > On Wed, Nov 15, 2017 at 5:07 PM, Romain Manni-Bucau < > >> >> rmannibu...@gmail.com> > >> >> > wrote: > >> >> > > >> >> >> It sounds a good start. I'm not sure how a group by key (and not > by > >> >> >> size) can help controlling the checkpointing interval. Wonder if > we > >> >> >> shouldn't be able to have a CheckpointPolicy { boolean > >> >> >> shouldCheckpoint() } used in the processing event loop. Default > could > >> >> >> be up to the runner but if set on the transform (or dofn) it > would be > >> >> >> used to control when the checkpoint is done. Thinking out loud it > >> >> >> sounds close to jbatch checkpoint algorithm > >> >> >> (https://docs.oracle.com/javaee/7/api/javax/batch/api/ > >> >> >> chunk/CheckpointAlgorithm.html) > >> >> >> > >> >> >> Romain Manni-Bucau > >> >> >> @rmannibucau | Blog | Old Blog | Github | LinkedIn > >> >> >> > >> >> >> > >> >> >> 2017-11-15 9:55 GMT+01:00 Jean-Baptiste Onofré <j...@nanthrax.net>: > >> >> >> > Yes, @StableReplay, that's the annotation. Thanks. > >> >> >> > > >> >> >> > > >> >> >> > On 11/15/2017 09:52 AM, Reuven Lax wrote: > >> >> >> >> > >> >> >> >> Romain, > >> >> >> >> > >> >> >> >> I think the @StableReplay semantic that Kenn proposed a month > or > >> so > >> >> ago > >> >> >> is > >> >> >> >> what is needed here. > >> >> >> >> > >> >> >> >> Essentially it will ensure that the GroupByKey iterable is > stable > >> and > >> >> >> >> checkpointed. So on replay, the GroupByKey is guaranteed to > >> receive > >> >> the > >> >> >> >> exact same iterable as it did before. The annotation can be put > >> on a > >> >> >> ParDo > >> >> >> >> as well, in which case it ensures stability (and > checkpointing) of > >> >> the > >> >> >> >> individual ParDo elements. > >> >> >> >> > >> >> >> >> Reuven > >> >> >> >> > >> >> >> >> On Wed, Nov 15, 2017 at 4:49 PM, Romain Manni-Bucau > >> >> >> >> <rmannibu...@gmail.com> > >> >> >> >> wrote: > >> >> >> >> > >> >> >> >>> 2017-11-15 9:46 GMT+01:00 Jean-Baptiste Onofré < > j...@nanthrax.net > >> >: > >> >> >> >>>> > >> >> >> >>>> Hi Romain, > >> >> >> >>>> > >> >> >> >>>> You are right: currently, the chunking is related to bundles. > >> >> Today, > >> >> >> the > >> >> >> >>>> bundle size is under the runner responsibility. > >> >> >> >>>> > >> >> >> >>>> I think it's fine because only the runner know an efficient > >> bundle > >> >> >> size. > >> >> >> >>> > >> >> >> >>> I'm > >> >> >> >>>> > >> >> >> >>>> afraid giving the "control" of the bundle size to the end > user > >> (via > >> >> >> >>>> pipeline) can result to huge performances issue depending of > the > >> >> >> runner. > >> >> >> >>>> > >> >> >> >>>> It doesn't mean that we can't use an uber layer: it's what we > >> do in > >> >> >> >>>> ParDoWithBatch or DoFn in IO Sink where we have a batch size. > >> >> >> >>>> > >> >> >> >>>> Anyway, the core problem is about the checkpoint: why a > >> checkpoint > >> >> is > >> >> >> >>>> not > >> >> >> >>>> "respected" by an IO or runner ? > >> >> >> >>> > >> >> >> >>> > >> >> >> >>> > >> >> >> >>> Take the example of a runner deciding the bundle size is 4 and > >> the > >> >> IO > >> >> >> >>> deciding the commit-interval (batch semantic) is 2, what > happens > >> if > >> >> >> >>> the 3rd record fails? You have pushed to the store 2 records > >> which > >> >> can > >> >> >> >>> be reprocessed by a restart of the bundle and you can get > >> >> duplicates. > >> >> >> >>> > >> >> >> >>> Rephrased: I think we need as a framework a batch/chunk > solution > >> >> which > >> >> >> >>> is reliable. I understand bundles is mapped on the runner and > not > >> >> >> >>> really controlled but can we get something more reliable for > the > >> >> user? > >> >> >> >>> Maybe we need a @BeforeBatch or something like that. > >> >> >> >>> > >> >> >> >>>> > >> >> >> >>>> Regards > >> >> >> >>>> JB > >> >> >> >>>> > >> >> >> >>>> > >> >> >> >>>> On 11/15/2017 09:38 AM, Romain Manni-Bucau wrote: > >> >> >> >>>>> > >> >> >> >>>>> > >> >> >> >>>>> Hi guys, > >> >> >> >>>>> > >> >> >> >>>>> The subject is a bit provocative but the topic is real and > >> coming > >> >> >> >>>>> again and again with the beam usage: how a dofn can handle > some > >> >> >> >>>>> "chunking". > >> >> >> >>>>> > >> >> >> >>>>> The need is to be able to commit each N records but with N > not > >> too > >> >> >> big. > >> >> >> >>>>> > >> >> >> >>>>> The natural API for that in beam is the bundle one but > bundles > >> are > >> >> >> not > >> >> >> >>>>> reliable since they can be very small (flink) - we can say > it > >> is > >> >> "ok" > >> >> >> >>>>> even if it has some perf impacts - or too big (spark does > full > >> >> size / > >> >> >> >>>>> #workers). > >> >> >> >>>>> > >> >> >> >>>>> The workaround is what we see in the ES I/O: a maxSize which > >> does > >> >> an > >> >> >> >>>>> eager flush. The issue is that then the checkpoint is not > >> >> respected > >> >> >> >>>>> and you can process multiple times the same records. > >> >> >> >>>>> > >> >> >> >>>>> Any plan to make this API reliable and controllable from a > beam > >> >> point > >> >> >> >>>>> of view (at least in a max manner)? > >> >> >> >>>>> > >> >> >> >>>>> Thanks, > >> >> >> >>>>> Romain Manni-Bucau > >> >> >> >>>>> @rmannibucau | Blog | Old Blog | Github | LinkedIn > >> >> >> >>>>> > >> >> >> >>>> > >> >> >> >>>> -- > >> >> >> >>>> Jean-Baptiste Onofré > >> >> >> >>>> jbono...@apache.org > >> >> >> >>>> http://blog.nanthrax.net > >> >> >> >>>> Talend - http://www.talend.com > >> >> >> >>> > >> >> >> >>> > >> >> >> >> > >> >> >> > > >> >> >> > -- > >> >> >> > Jean-Baptiste Onofré > >> >> >> > jbono...@apache.org > >> >> >> > http://blog.nanthrax.net > >> >> >> > Talend - http://www.talend.com > >> >> >> > >> >> > >> >