public <InputT, CollectionT extends PCollection<? extends InputT>>
      Map<String, DataSet<?>> applyMultiOutputBeamPTransform(
          DataSet<InputT> input, PTransform<CollectionT, PCollectionTuple> transform) {
    return applyBeamPTransformInternal(
        ImmutableMap.of("input", input),
        (pipeline, map) -> (CollectionT) getNonNull(map, "input"),
        BeamAdapterUtils::tupleToMap,
        transform,
        input.getExecutionEnvironment());
  }
@Test
  public void testApplyMultiOutputTransform() throws Exception {
    ExecutionEnvironment env = ExecutionEnvironment.createCollectionsEnvironment();

    DataSet<String> input = env.fromCollection(ImmutableList.of("a", "b", "c"));
    Map<String, DataSet<?>> result =
        new BeamFlinkDataSetAdapter()
            .applyMultiOutputBeamPTransform(
                input,
                new PTransform<PCollection<String>, PCollectionTuple>() {
                  @Override
                  public PCollectionTuple expand(PCollection<String> input) {
                    return PCollectionTuple.of("x", input.apply(withPrefix("x")))
                        .and("y", input.apply(withPrefix("y")));
                  }
                });

    assertThat(result.get("x").collect(), containsInAnyOrder("xa", "xb", "xc"));
    assertThat(result.get("y").collect(), containsInAnyOrder("ya", "yb", "yc"));
  }