public static <InputT extends PInput, OutputT extends POutput>
      PTransform<InputT, OutputT> compose(SerializableFunction<InputT, OutputT> fn) {
    return new PTransform<InputT, OutputT>() {
      @Override
      public OutputT expand(InputT input) {
        return fn.apply(input);
      }
    };
  }
@Test
  public void testNamedCompose() {
    PTransform<PCollection<Integer>, PCollection<Integer>> composed =
        PTransform.compose("MyName", (PCollection<Integer> numbers) -> numbers);
    assertEquals("MyName", composed.name);
  }