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
