Window extends PTransform<PCollection<T>, PCollection<T>> { public static <T> Window<T> into(WindowFn<? super T, ?> fn) { try { fn.windowCoder().verifyDeterministic(); } catch (NonDeterministicException e) { throw new IllegalArgumentException("Window coders must be deterministic.", e); } return Window.<T>configure().withWindowFn(fn); }  static Window<T> into(WindowFn<? super T, ?> fn); static Window<T> configure(); abstract @Nullable WindowFn<? super T, ?> getWindowFn(); @Experimental(Kind.TRIGGER) Window<T> triggering(Trigger trigger); @Experimental(Kind.TRIGGER) Window<T> discardingFiredPanes(); @Experimental(Kind.TRIGGER) Window<T> accumulatingFiredPanes(); @Experimental(Kind.TRIGGER) Window<T> withAllowedLateness(Duration allowedLateness); @Experimental(Kind.OUTPUT_TIME) Window<T> withTimestampCombiner(TimestampCombiner timestampCombiner); @Experimental(Kind.TRIGGER) Window<T> withAllowedLateness(Duration allowedLateness, ClosingBehavior behavior); @Experimental(Kind.TRIGGER) Window<T> withOnTimeBehavior(OnTimeBehavior behavior); WindowingStrategy<?, ?> getOutputStrategyInternal(WindowingStrategy<?, ?> inputStrategy); @Override PCollection<T> expand(PCollection<T> input); @Override void populateDisplayData(DisplayData.Builder builder); static Remerge<T> remerge();  }
@Test public void testWindowIntoWindowFnAssign() { pipeline .apply(Create.of(1, 2, 3)) .apply( Window.into(FixedWindows.of(Duration.standardMinutes(11L).plus(Duration.millis(1L))))); final AtomicBoolean foundAssign = new AtomicBoolean(false); pipeline.traverseTopologically( new PipelineVisitor.Defaults() { @Override public void visitPrimitiveTransform(TransformHierarchy.Node node) { if (node.getTransform() instanceof Window.Assign) { foundAssign.set(true); } } }); assertThat(foundAssign.get(), is(true)); }
@Test public void testWindowGetName() { assertEquals( "Window.Into()", Window.<String>into(FixedWindows.of(Duration.standardMinutes(10))).getName()); }
@Test public void testNonDeterministicWindowCoder() throws NonDeterministicException { FixedWindows mockWindowFn = Mockito.mock(FixedWindows.class); @SuppressWarnings({"unchecked", "rawtypes"}) Class<Coder<IntervalWindow>> coderClazz = (Class) Coder.class; Coder<IntervalWindow> mockCoder = Mockito.mock(coderClazz); when(mockWindowFn.windowCoder()).thenReturn(mockCoder); NonDeterministicException toBeThrown = new NonDeterministicException(mockCoder, "Its just not deterministic."); Mockito.doThrow(toBeThrown).when(mockCoder).verifyDeterministic(); thrown.expect(IllegalArgumentException.class); thrown.expectCause(Matchers.sameInstance(toBeThrown)); thrown.expectMessage("Window coders must be deterministic"); Window.into(mockWindowFn); }
@Test @Category(ValidatesRunner.class) public void testTimestampCombinerDefault() { pipeline.enableAbandonedNodeEnforcement(true); pipeline .apply( Create.timestamped( TimestampedValue.of(KV.of(0, "hello"), new Instant(0)), TimestampedValue.of(KV.of(0, "goodbye"), new Instant(10)))) .apply(Window.into(FixedWindows.of(Duration.standardMinutes(10)))) .apply(GroupByKey.create()) .apply( ParDo.of( new DoFn<KV<Integer, Iterable<String>>, Void>() { @ProcessElement public void processElement(ProcessContext c) throws Exception { assertThat( c.timestamp(), equalTo( new IntervalWindow( new Instant(0), new Instant(0).plus(Duration.standardMinutes(10))) .maxTimestamp())); } })); pipeline.run(); }
@Test @Category({ValidatesRunner.class, UsesCustomWindowMerging.class}) public void testMergingCustomWindows() { Instant startInstant = new Instant(0L); PCollection<String> inputCollection = pipeline.apply( Create.timestamped( TimestampedValue.of("big", startInstant.plus(Duration.standardSeconds(10))), TimestampedValue.of("small1", startInstant.plus(Duration.standardSeconds(20))), TimestampedValue.of("small2", startInstant.plus(Duration.standardSeconds(39))))); PCollection<String> windowedCollection = inputCollection.apply(Window.into(new CustomWindowFn<>())); PCollection<Long> count = windowedCollection.apply(Combine.globally(Count.<String>combineFn()).withoutDefaults()); PAssert.that("Wrong number of elements in output collection", count).containsInAnyOrder(2L, 1L); pipeline.run(); }
@Test @Category({ValidatesRunner.class, UsesCustomWindowMerging.class}) public void testMergingCustomWindowsKeyedCollection() { Instant startInstant = new Instant(0L); PCollection<KV<Integer, String>> inputCollection = pipeline.apply( Create.timestamped( TimestampedValue.of( KV.of(0, "big"), startInstant.plus(Duration.standardSeconds(10))), TimestampedValue.of( KV.of(1, "small1"), startInstant.plus(Duration.standardSeconds(20))), TimestampedValue.of( KV.of(2, "small2"), startInstant.plus(Duration.standardSeconds(39))))); PCollection<KV<Integer, String>> windowedCollection = inputCollection.apply(Window.into(new CustomWindowFn<>())); PCollection<Long> count = windowedCollection.apply( Combine.globally(Count.<KV<Integer, String>>combineFn()).withoutDefaults()); PAssert.that("Wrong number of elements in output collection", count).containsInAnyOrder(2L, 1L); pipeline.run(); }