PTransform implements Serializable /* See the note above */, HasDisplayData { @Experimental 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); } }; } protected  PTransform(); protected  PTransform(@Nullable String name); abstract OutputT expand(InputT input); void validate(@Nullable PipelineOptions options); Map<TupleTag<?>, PValue> getAdditionalInputs(); String getName(); @Override String toString(); @Deprecated Coder<T> getDefaultOutputCoder(
      InputT input, @SuppressWarnings("unused") PCollection<T> output); @Override void populateDisplayData(Builder builder); @Experimental static PTransform<InputT, OutputT> compose(SerializableFunction<InputT, OutputT> fn); @Experimental static PTransform<InputT, OutputT> compose(String name, SerializableFunction<InputT, OutputT> fn);  }
@Test public void testNamedCompose() { PTransform<PCollection<Integer>, PCollection<Integer>> composed = PTransform.compose("MyName", (PCollection<Integer> numbers) -> numbers); assertEquals("MyName", composed.name); }
@Test @Category(NeedsRunner.class) public void testComposeBasicSerializableFunction() throws Exception { PCollection<Integer> output = pipeline .apply(Create.of(1, 2, 3)) .apply( PTransform.compose( (PCollection<Integer> numbers) -> { PCollection<Integer> inverted = numbers.apply(MapElements.into(integers()).via(input -> -input)); return PCollectionList.of(numbers) .and(inverted) .apply(Flatten.pCollections()); })); PAssert.that(output).containsInAnyOrder(-2, -1, -3, 2, 1, 3); pipeline.run(); }