I'm in the process of writing a simple test to verify the semantics of early/on-time/late panes. The pipeline combines the number of elements per key. My early and on-time panes are working as expected, though my final pane seems to be empty at all times.
private static final Duration WINDOW_LENGTH = Duration.standardMinutes(2);
private static final Duration LATENESS_HORIZON = Duration.standardDays(1);
My test is as follows:
@Test
@Category(ValidatesRunner.class)
public void simpleTest() throws Exception {
Instant baseTime = new Instant(0L);
Duration one_min = Duration.standardMinutes(1);
TestStream<KV<String, Long>> events = TestStream.create(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()))
.advanceWatermarkTo(baseTime)
// First element arrives
.addElements(
TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
)
.advanceProcessingTime(Duration.standardMinutes(5))
// Second element arrives
.addElements(
TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
)
.advanceProcessingTime(Duration.standardMinutes(5))
// Third element arrives
.addElements(
TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
)
.advanceProcessingTime(Duration.standardMinutes(5))
// Window ends
.advanceWatermarkTo(baseTime.plus(WINDOW_LENGTH).plus(one_min))
// Late element arrives
.addElements(
TimestampedValue.of(KV.of("laurens", 0L), baseTime.plus(one_min))
)
.advanceProcessingTime(Duration.standardMinutes(5))
// Fire all
.advanceWatermarkToInfinity();
PCollection<KV<String, Long>> userCount = p.apply(events).apply(new CountPipeline());
IntervalWindow window = new IntervalWindow(baseTime, WINDOW_LENGTH);
PAssert.that(userCount) // This test works
.inEarlyPane(window)
.containsInAnyOrder(
KV.of("laurens", 1L), // First firing
KV.of("laurens", 2L), // Second firing
KV.of("laurens", 3L) // Third firing
);
PAssert.that(userCount) // This test works as well
.inOnTimePane(window)
.containsInAnyOrder(
KV.of("laurens", 3L) // On time firing
);
PAssert.that(userCount) // Test fails
.inFinalPane(window)
.containsInAnyOrder(
KV.of("laurens", 4L) // Late firing
);
p.run().waitUntilFinish();
}
Code of the pipeline is as follows:
public static class CountPipeline extends PTransform<PCollection<KV<String, Long>>, PCollection<KV<String, Long>>> {
@Override
public PCollection<KV<String, Long>> expand(PCollection<KV<String, Long>> events) {
return events.apply("window", Window.<KV<String, Long>>into(FixedWindows.of(WINDOW_LENGTH))
.triggering(AfterWatermark
.pastEndOfWindow()
.withEarlyFirings(AfterProcessingTime
.pastFirstElementInPane())
.withLateFirings(AfterProcessingTime
.pastFirstElementInPane())
)
.withAllowedLateness(LATENESS_HORIZON)
.accumulatingFiredPanes()
.withOnTimeBehavior(Window.OnTimeBehavior.FIRE_ALWAYS)
).apply("Count", Count.perKey());
}
}
Error:
Expected: iterable over [<KV{laurens, 4}>] in any order
but: No item matches: <KV{laurens, 4}> in []
As you can see, the last element is definitely ingested after the watermark, which should make it late by definition. Though, the final pane does not contain a refinement of the original result. I'm honestly clueless why it the late pane is not firing. Any insights are appreciated.
Aucun commentaire:
Enregistrer un commentaire