Count { public static <T> PTransform<PCollection<T>, PCollection<KV<T, Long>>> perElement() { return new PerElement<>(); } private  Count(); static CombineFn<T, ?, Long> combineFn(); static PTransform<PCollection<T>, PCollection<Long>> globally(); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, Long>>> perKey(); static PTransform<PCollection<T>, PCollection<KV<T, Long>>> perElement();  }
@Test @Category(NeedsRunner.class) @SuppressWarnings("unchecked") public void testCountPerElementBasic() { PCollection<String> input = p.apply(Create.of(WORDS)); PCollection<KV<String, Long>> output = input.apply(Count.perElement()); PAssert.that(output) .containsInAnyOrder( KV.of("hi", 4L), KV.of("there", 1L), KV.of("sue", 2L), KV.of("bob", 2L), KV.of("", 3L), KV.of("ZOW", 1L)); p.run(); }
@Test @Category(NeedsRunner.class) @SuppressWarnings("unchecked") public void testCountPerElementEmpty() { PCollection<String> input = p.apply(Create.of(NO_LINES).withCoder(StringUtf8Coder.of())); PCollection<KV<String, Long>> output = input.apply(Count.perElement()); PAssert.that(output).empty(); p.run(); }