osdir.com

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Generating a window identifier while using a Trigger AfterProcessingTime.pastFirstElementInPane()


I looked into the PaneInfo but unfortunately it doesn't contain information regarding the session window that it is useful for me (at least to tags all the events belonging to same session with a  sessionID) . :(  

I have no idea of BEAM internal or how complex would be to implement it, but it would be cool if such hook or information (windowIdentifier) exist.

Thanks!

On Thu, Jul 19, 2018 at 10:18 PM, Lukasz Cwik <lcwik@xxxxxxxxxx> wrote:
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 <jcgarciam@xxxxxxxxx> 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 




--

JC