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