public static <T> PTransform<PCollection<T>, PCollection<KV<T, Long>>> perElement() {
    return new 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();
  }