import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
private static <T> TransformEvaluator<Flatten.FlattenPCollectionList<T>> flattenPColl() {
  return new TransformEvaluator<Flatten.FlattenPCollectionList<T>>() {
    @SuppressWarnings("unchecked")
    @Override
    public void evaluate(Flatten.FlattenPCollectionList<T> transform, EvaluationContext context) {
      StreamingEvaluationContext sec = (StreamingEvaluationContext) context;
      PCollectionList<T> pcs = sec.getInput(transform);
      JavaDStream<WindowedValue<T>> first =
          (JavaDStream<WindowedValue<T>>) sec.getStream(pcs.get(0));
      List<JavaDStream<WindowedValue<T>>> rest = Lists.newArrayListWithCapacity(pcs.size() - 1);
      for (int i = 1; i < pcs.size(); i++) {
        rest.add((JavaDStream<WindowedValue<T>>) sec.getStream(pcs.get(i)));
      }
      JavaDStream<WindowedValue<T>> dstream = sec.getStreamingContext().union(first, rest);
      sec.setStream(transform, dstream);
    }
  };
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
@Override
public PCollection<KV<String, String>> apply(PInput input) {
  Pipeline pipeline = input.getPipeline();

  // Create one TextIO.Read transform for each data file
  // and add its output to a PCollectionList.
  PCollectionList<KV<String, String>> filesToLines = PCollectionList.empty(pipeline);

  for (final String fileLocation : files) {
    PTransform<PInput, PCollection<String>> inputSource
        = TextIO.Read.from(fileLocation)
            .named("TextIO.Read(" + fileLocation + ")");

    PCollection<KV<String, String>> oneFileToLines = pipeline
        .apply(inputSource)
        .apply(WithKeys.<String, String>of(fileLocation));

    filesToLines = filesToLines.and(oneFileToLines);
  }

  return filesToLines.apply(Flatten.<KV<String, String>> pCollections())
                     .setCoder(getDefaultOutputCoder());
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
@Override
public PCollection<KV<String, WorkflowArgs>> apply(
    PCollectionList<KV<String, WorkflowArgs>> input) {
  return input
      .apply(Flatten.<KV<String, WorkflowArgs>>pCollections())
      .apply(Combine.globally(new Merge()));
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
public PCollection<KV<String, TSProto>> generateCompleteWindowData(Pipeline pipeline,
    List<KV<String, TSProto>> data, WorkPacketConfig packetConfig) {

  LOG.info("Check to see that time streams with missing 'ticks' have been corrected");

  PCollection<KV<String, TSProto>> tsData = setupDataInput(pipeline, data);


  PCollection<KV<String, TSProto>> windowedData =
      tsData.apply("CandleResolutionWindow", Window.<KV<String, TSProto>>into(FixedWindows
          .of(Duration.standardSeconds(((FXTimeSeriesPipelineOptions) pipeline.getOptions())
              .getCandleResolution()))));

  // Determine streams that are missing in this Window and generate values for them

  PCollection<KV<String, TSProto>> generatedValues =
      windowedData
          .apply(
              "DetectMissingTimeSeriesValues",
              Combine.globally(new DetectMissingTimeSeriesValuesCombiner(packetConfig))
                  .withoutDefaults()).apply(ParDo.of(new CreateMissingTimeSeriesValuesDoFn()))
          .setName("CreateMissingTimeSeriesValues");

  // Flatten the live streams and the generated streams together

  PCollection<KV<String, TSProto>> completeWindowData =
      PCollectionList.of(windowedData).and(generatedValues)
          .apply("MergeGeneratedLiveValues", Flatten.<KV<String, TSProto>>pCollections());


  return completeWindowData;
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
private static <T> TransformEvaluator<Flatten.FlattenPCollectionList<T>> flattenPColl() {
  return new TransformEvaluator<Flatten.FlattenPCollectionList<T>>() {
    @SuppressWarnings("unchecked")
    @Override
    public void evaluate(Flatten.FlattenPCollectionList<T> transform, EvaluationContext context) {
      PCollectionList<T> pcs = context.getInput(transform);
      JavaRDD<WindowedValue<T>>[] rdds = new JavaRDD[pcs.size()];
      for (int i = 0; i < rdds.length; i++) {
        rdds[i] = (JavaRDD<WindowedValue<T>>) context.getRDD(pcs.get(i));
      }
      JavaRDD<WindowedValue<T>> rdd = context.getSparkContext().union(rdds);
      context.setOutputRDD(transform, rdd);
    }
  };
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
@Test
public void testRun() throws Exception {
  Pipeline p = Pipeline.create(PipelineOptionsFactory.create());
  PCollection<String> regex = p.apply(Create.of("[^a-zA-Z']+"));
  PCollection<String> w1 = p.apply(Create.of("Here are some words to count", "and some others"));
  PCollection<String> w2 = p.apply(Create.of("Here are some more words", "and even more words"));
  PCollectionList<String> list = PCollectionList.of(w1).and(w2);

  PCollection<String> union = list.apply(Flatten.<String>pCollections());
  PCollectionView<String> regexView = regex.apply(View.<String>asSingleton());
  CountWords countWords = new CountWords(regexView);
  PCollectionTuple luc = union.apply(countWords);
  PCollection<Long> unique = luc.get(lowerCnts).apply(
      ApproximateUnique.<KV<String, Long>>globally(16));

  EvaluationResult res = SparkPipelineRunner.create().run(p);
  Iterable<KV<String, Long>> actualLower = res.get(luc.get(lowerCnts));
  Assert.assertEquals("are", actualLower.iterator().next().getKey());
  Iterable<KV<String, Long>> actualUpper = res.get(luc.get(upperCnts));
  Assert.assertEquals("Here", actualUpper.iterator().next().getKey());
  Iterable<Long> actualUniqCount = res.get(unique);
  Assert.assertEquals(9, (long) actualUniqCount.iterator().next());
  int actualTotalWords = res.getAggregatorValue("totalWords", Integer.class);
  Assert.assertEquals(18, actualTotalWords);
  int actualMaxWordLength = res.getAggregatorValue("maxWordLength", Integer.class);
  Assert.assertEquals(6, actualMaxWordLength);
  AggregatorValues<Integer> aggregatorValues = res.getAggregatorValues(countWords
      .getTotalWordsAggregator());
  Assert.assertEquals(18, Iterables.getOnlyElement(aggregatorValues.getValues()).intValue());

  res.close();
}
 

import com.google.cloud.dataflow.sdk.transforms.Flatten; //导入依赖的package包/类
@Test
public void testRun() throws Exception {
  SparkStreamingPipelineOptions options = SparkStreamingPipelineOptionsFactory.create();
  options.setAppName(this.getClass().getSimpleName());
  options.setRunner(SparkPipelineRunner.class);
  options.setTimeout(TEST_TIMEOUT_MSEC);// run for one interval
  Pipeline p = Pipeline.create(options);

  PCollection<String> w1 =
          p.apply(CreateStream.fromQueue(WORDS_QUEUE_1)).setCoder(StringUtf8Coder.of());
  PCollection<String> windowedW1 =
          w1.apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(1))));
  PCollection<String> w2 =
          p.apply(CreateStream.fromQueue(WORDS_QUEUE_2)).setCoder(StringUtf8Coder.of());
  PCollection<String> windowedW2 =
          w2.apply(Window.<String>into(FixedWindows.of(Duration.standardSeconds(1))));
  PCollectionList<String> list = PCollectionList.of(windowedW1).and(windowedW2);
  PCollection<String> union = list.apply(Flatten.<String>pCollections());

  DataflowAssert.thatIterable(union.apply(View.<String>asIterable()))
          .containsInAnyOrder(EXPECTED_UNION);

  EvaluationResult res = SparkPipelineRunner.create(options).run(p);
  res.close();

  DataflowAssertStreaming.assertNoFailures(res);
}