public static <T> PTransform<PCollection<T>, PCollection<Long>> globally() {
    return Combine.globally(new CountFn<T>());
  }
@Test
  @Category(NeedsRunner.class)
  public void testCountGloballyEmpty() {
    PCollection<String> input = p.apply(Create.of(NO_LINES).withCoder(StringUtf8Coder.of()));

    PCollection<Long> output = input.apply(Count.globally());

    PAssert.that(output).containsInAnyOrder(0L);
    p.run();
  }