public <InputT, OutputT, CollectionT extends PCollection<? extends InputT>>
      DataStream<OutputT> applyBeamPTransform(
          DataStream<InputT> input, PTransform<CollectionT, PCollection<OutputT>> transform) {
    return (DataStream)
        getNonNull(
            applyBeamPTransformInternal(
                ImmutableMap.of("input", input),
                (pipeline, map) -> (CollectionT) getNonNull(map, "input"),
                (output) -> ImmutableMap.of("output", output),
                transform,
                input.getExecutionEnvironment()),
            "output");
  }
@Test
  public void testApplyGroupingTransform() throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

    DataStream<String> input = env.fromCollection(ImmutableList.of("a", "a", "b"));
    DataStream<KV<String, Long>> result =
        new BeamFlinkDataStreamAdapter()
            .applyBeamPTransform(
                input,
                new PTransform<PCollection<String>, PCollection<KV<String, Long>>>() {
                  @Override
                  public PCollection<KV<String, Long>> expand(PCollection<String> input) {
                    return input
                        .apply(Window.into(FixedWindows.of(Duration.millis(10))))
                        .apply(Count.perElement());
                  }
                });

    assertThat(
        ImmutableList.copyOf(result.executeAndCollect()),
        containsInAnyOrder(KV.of("a", 2L), KV.of("b", 1L)));
  }