osdir.com

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

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


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