MultiStepCombine extends RawPTransform<PCollection<KV<K, InputT>>, PCollection<KV<K, OutputT>>> { public static < K extends @Nullable Object, InputT extends @Nullable Object, AccumT extends @Nullable Object, OutputT extends @Nullable Object> MultiStepCombine<K, InputT, AccumT, OutputT> of( CombineFn<InputT, AccumT, OutputT> combineFn, Coder<KV<K, OutputT>> outputCoder) { return new MultiStepCombine<>(combineFn, outputCoder); } private  MultiStepCombine(
      CombineFn<InputT, AccumT, OutputT> combineFn, Coder<KV<K, OutputT>> outputCoder); static PTransformMatcher matcher(); static MultiStepCombine<K, InputT, AccumT, OutputT> of(
          CombineFn<InputT, AccumT, OutputT> combineFn, Coder<KV<K, OutputT>> outputCoder); @Override @NonNull String getUrn(); @Override @Nullable FunctionSpec getSpec(); @Override PCollection<KV<K, OutputT>> expand(PCollection<KV<K, InputT>> input);  }
@Test public void testMultiStepCombineWindowed() { SlidingWindows windowFn = SlidingWindows.of(Duration.millis(6L)).every(Duration.millis(3L)); PCollection<KV<String, Long>> combined = pipeline .apply( Create.timestamped( TimestampedValue.of(KV.of("foo", 1L), new Instant(1L)), TimestampedValue.of(KV.of("bar", 2L), new Instant(2L)), TimestampedValue.of(KV.of("bizzle", 3L), new Instant(3L)), TimestampedValue.of(KV.of("bar", 4L), new Instant(4L)), TimestampedValue.of(KV.of("bizzle", 11L), new Instant(11L)))) .apply(Window.into(windowFn)) .apply(Combine.perKey(new MultiStepCombineFn())); PAssert.that("Windows should combine only elements in their windows", combined) .inWindow(new IntervalWindow(new Instant(0L), Duration.millis(6L))) .containsInAnyOrder(KV.of("foo", 1L), KV.of("bar", 6L), KV.of("bizzle", 3L)); PAssert.that("Elements should appear in all the windows they are assigned to", combined) .inWindow(new IntervalWindow(new Instant(-3L), Duration.millis(6L))) .containsInAnyOrder(KV.of("foo", 1L), KV.of("bar", 2L)); PAssert.that(combined) .inWindow(new IntervalWindow(new Instant(6L), Duration.millis(6L))) .containsInAnyOrder(KV.of("bizzle", 11L)); PAssert.that(combined) .containsInAnyOrder( KV.of("foo", 1L), KV.of("foo", 1L), KV.of("bar", 6L), KV.of("bar", 2L), KV.of("bar", 4L), KV.of("bizzle", 11L), KV.of("bizzle", 11L), KV.of("bizzle", 3L), KV.of("bizzle", 3L)); pipeline.run(); }
@Test public void testMultiStepCombineTimestampCombiner() { TimestampCombiner combiner = TimestampCombiner.LATEST; combinedCoder = KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()); PCollection<KV<String, Long>> combined = pipeline .apply( Create.timestamped( TimestampedValue.of(KV.of("foo", 4L), new Instant(1L)), TimestampedValue.of(KV.of("foo", 1L), new Instant(4L)), TimestampedValue.of(KV.of("bazzle", 4L), new Instant(4L)), TimestampedValue.of(KV.of("foo", 12L), new Instant(12L)))) .apply( Window.<KV<String, Long>>into(FixedWindows.of(Duration.millis(5L))) .withTimestampCombiner(combiner)) .apply(Combine.perKey(new MultiStepCombineFn())); PCollection<KV<String, TimestampedValue<Long>>> reified = combined.apply( ParDo.of( new DoFn<KV<String, Long>, KV<String, TimestampedValue<Long>>>() { @ProcessElement public void reifyTimestamp(ProcessContext context) { context.output( KV.of( context.element().getKey(), TimestampedValue.of( context.element().getValue(), context.timestamp()))); } })); PAssert.that(reified) .containsInAnyOrder( KV.of("foo", TimestampedValue.of(5L, new Instant(4L))), KV.of("bazzle", TimestampedValue.of(4L, new Instant(4L))), KV.of("foo", TimestampedValue.of(12L, new Instant(12L)))); pipeline.run(); }