Note that @StartBundle is not co-related with a new pane, but an arbitrary runner chosen amount of elements containing any number of windows and elements restricted by the triggering semantics.
You can introspect the PaneInfo to see the firing index, index 0 represents the first firing. I don't believe there is a way to know what is the last firing without using a trigger that will produce a known number of firings (e.g. watermark trigger with no speculative or late firings). On Thu, Jul 19, 2018 at 6:12 AM Juan Carlos Garcia <[email protected]> wrote: > Hi Folks, > > I would like to ask if its possible to be notified when a Windows is > created or closed while processing a batch of data. (Sorry for the long > post) > > My scenario: > I am using a Session window with a GapDuration of 2 minutes (for testing), > during this processing we are assigning a Session identifier to the > incoming messages so we can identify them later in ElasticSearch / Other > tools, the process works as expected as long as we don't introduce any > trigger (during the @ProcessElement we have the the Iterables elements from > this windows and from there we can just generate our session identifier > like) i.e: > > <code> > PCollection<KV<Long, Iterable<String>>> windowedResult = input > .apply("Session", > Window.<String>into(Sessions.withGapDuration(Duration.standardMinutes(2)))) > .apply("Create KV of Users", ParDo.of(new CreateMyKV())) > .apply(GroupByKey.create()) > .apply(ParDo.of(new DoFn<KV<Long, Iterable<String>>, KV<Long, > Iterable<String>>>() { > @ProcessElement > public void processElement(ProcessContext c, BoundedWindow _window) { > System.out.println("-- window:" + _window); > System.out.println("session:" + UUID.randomUUID().toString()); > System.out.println(c.element().getValue()); > System.out.println("--"); > c.output(c.element()); > } > })); > </code> > > After i added the trigger "AfterProcessingTime.pastFirstElementInPane()", > each fired pane doesn't contain any indications of the windows they belong > to, and there is no way (at least i couldn't find) to actually hook into it > and generate a Session identifier for the elements that belongs to the same > windows. > > The behavior for @StartBundle is that it fires for each pane and the > behavior for @Setup is not consistent as it fires more times than windows > we have or sometime it fires less time. > > Any advised on this matter is welcome and by the way, in production we are > using the SparkRunner (which only support ProcessingTime triggers based on > the capability-matrix), please find below a JUnit class i am using to > validate this behavior. > > <code> > public class SessionWindowTest { > private long TIME = System.currentTimeMillis(); > > @Test > public void testSessionWindowWorkAsExpected() { > final List<String> testMesages = new LinkedList<>(); > TIME = System.currentTimeMillis(); > > // > // 3 Windows of data > // > > IntStream.range(0, 10).forEach(i -> { > testMesages.add("{\"user_id\":123456789, > \"event_type\":\"watch_tutorial\", \"time\":" + (TIME) + "}"); > TIME += TimeUnit.SECONDS.toMillis(1); > }); > TIME += TimeUnit.MINUTES.toMillis( 5); > IntStream.range(0, 10).forEach(i -> { > testMesages.add("{\"user_id\":123456789, > \"event_type\":\"watch_tutorial\", \"time\":" + (TIME) + "}"); > TIME += TimeUnit.SECONDS.toMillis(2); > }); > TIME += TimeUnit.MINUTES.toMillis(6); > IntStream.range(0, 10).forEach(i -> { > testMesages.add("{\"user_id\":123456789, > \"event_type\":\"watch_tutorial\", \"time\":" + (TIME) + "}"); > TIME += TimeUnit.SECONDS.toMillis(4); > }); > > Pipeline pipe = Pipeline.create(); > PCollection<String> input = pipe.apply("Create", > Create.of(testMesages)); > > PCollection<KV<Long, Iterable<String>>> windowedResult = > input.apply("setting the time", > ParDo.of(new DoFn<String, String>() { > @ProcessElement > public void processElement(ProcessContext c) { > Any deserialize = > JsonIterator.deserialize(c.element()); > c.outputWithTimestamp(c.element(), new > Instant(deserialize.get("time").toLong())); > } > })) > .apply("Session", > > Window.<String>into(Sessions.withGapDuration(Duration.standardMinutes(2))) > // > .withAllowedLateness(Duration.standardSeconds(1)) > // .discardingFiredPanes() > // > .triggering(AfterProcessingTime.pastFirstElementInPane()) > ) > .apply("Create KV of Users", ParDo.of(new CreateUserKV())) > .apply(GroupByKey.create()) > .apply(ParDo.of(new DoFn<KV<Long, Iterable<String>>, > KV<Long, Iterable<String>>>() { > private int counter = 0; > > @StartBundle > public void startBundle() { > System.out.println("--BUNDLE--"); > } > > @Setup > public void setupFn() { > System.out.println("--SETUP--"); > } > @ProcessElement > public void processElement(ProcessContext c, > BoundedWindow _window) { > System.out.println("-- window:" + _window); > System.out.println("session:" + > UUID.randomUUID().toString()); > //System.out.println(c.element().getValue()); > System.out.println("--"); > c.output(c.element()); > } > })); > > PipelineResult run = pipe.run(); > Assert.assertTrue("Pipeline is done", run.getState() == > PipelineResult.State.DONE); > } > > static class CreateMyKV extends DoFn<String, KV<Long, String>> { > @ProcessElement > public void processElement(ProcessContext c) { > Any deserialize = JsonIterator.deserialize(c.element()); > Map<String, Any> stringAnyMap = deserialize.asMap(); > c.output(KV.of(stringAnyMap.get("user_id").toLong(), > c.element())); > } > } > } > </code> > > -- > > JC > >
