import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
private Pipeline setupPipeline(final String inputPath, final String outputPath, boolean enableGcs, boolean enableCloudExec) {
    final GATKGCSOptions options = PipelineOptionsFactory.as(GATKGCSOptions.class);
    if (enableCloudExec) {
        options.setStagingLocation(getGCPTestStaging());
        options.setProject(getGCPTestProject());
        options.setRunner(BlockingDataflowPipelineRunner.class);
    } else if (BucketUtils.isHadoopUrl(inputPath) || BucketUtils.isHadoopUrl(outputPath)) {
        options.setRunner(SparkPipelineRunner.class);
    } else {
        options.setRunner(DirectPipelineRunner.class);
    }
    if (enableGcs) {
        options.setApiKey(getGCPTestApiKey());
    }
    final Pipeline p = Pipeline.create(options);
    DataflowUtils.registerGATKCoders(p);
    return p;
}
 

import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
/**
 * Do some runner setup: check that the DirectPipelineRunner is not used in conjunction with
 * streaming, and if streaming is specified, use the DataflowPipelineRunner. Return the streaming
 * flag value.
 */
public void setupRunner() {
  if (options.isStreaming()) {
    if (options.getRunner() == DirectPipelineRunner.class) {
      throw new IllegalArgumentException(
        "Processing of unbounded input sources is not supported with the DirectPipelineRunner.");
    }
    // In order to cancel the pipelines automatically,
    // {@literal DataflowPipelineRunner} is forced to be used.
    options.setRunner(DataflowPipelineRunner.class);
  }
}
 

import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
@org.junit.Test
public void joinArtistCreditsWithRecordings() {

  DirectPipeline p = DirectPipeline.createForTest();

  PCollection<String> artistCreditText = p.apply("artistCredits", Create.of(artistCreditLinesOfJson)).setCoder(StringUtf8Coder.of());
  PCollection<KV<Long, MusicBrainzDataObject>> artistCredits = MusicBrainzTransforms.loadTableFromText(artistCreditText, "artist_credit_name", "artist_credit");

  PCollection<String> recordingText = p.apply("recordings", Create.of(recordingLinesOfJson)).setCoder(StringUtf8Coder.of());
  PCollection<KV<Long, MusicBrainzDataObject>> recordings = MusicBrainzTransforms.loadTableFromText(recordingText, "recording", "artist_credit");

  PCollection<MusicBrainzDataObject> joinedRecordings = MusicBrainzTransforms.innerJoin("artist credits with recordings", artistCredits, recordings);

  PCollection<Long> recordingIds = joinedRecordings.apply(MapElements.via((MusicBrainzDataObject mbo) -> (Long) mbo.getColumnValue("recording_id")).
      withOutputType(new TypeDescriptor<Long>() {
      }));

  Long bieberRecording = 17069165L;
  Long bieberRecording2 = 15508507L;


  DataflowAssert.that(recordingIds).satisfies((longs) -> {
    List<Long> theList = new ArrayList<Long>();
    longs.forEach(theList::add);
    assert (theList.contains(bieberRecording));
    assert (theList.contains(bieberRecording2));
    return null;
  });

  PCollection<Long> numberJoined = joinedRecordings.apply("count joined recrodings", Count.globally());
  PCollection<Long> numberOfArtistCredits = artistCredits.apply("count artist credits", Count.globally());

  DirectPipelineRunner.EvaluationResults results = p.run();

  long joinedRecordingsCount = results.getPCollection(numberJoined).get(0);
  assert (448 == joinedRecordingsCount);
}
 

import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
@org.junit.Test
public void testNest() {
  DirectPipeline p = DirectPipeline.createForTest();
  PCollection<String> artistText = p.apply("artist", Create.of(artistLinesOfJson)).setCoder(StringUtf8Coder.of());
  PCollection<String> artistCreditNameText = p.apply("artist_credit_name", Create.of(artistCreditLinesOfJson));
  PCollection<String> recordingText = p.apply("recording", Create.of(recordingLinesOfJson)).setCoder(StringUtf8Coder.of());

  PCollection<KV<Long, MusicBrainzDataObject>> artistsById = MusicBrainzTransforms.loadTableFromText(artistText, "artist", "id");

  PCollection<KV<Long, MusicBrainzDataObject>> recordingsByArtistCredit =
      MusicBrainzTransforms.loadTableFromText(recordingText, "recording", "artist_credit");
  PCollection<KV<Long, MusicBrainzDataObject>> artistCreditByArtistCredit =
      MusicBrainzTransforms.loadTableFromText(artistCreditNameText, "artist_credit_name", "artist_credit");

  PCollection<MusicBrainzDataObject> recordingsWithCredits = MusicBrainzTransforms.innerJoin("credited recordings", artistCreditByArtistCredit, recordingsByArtistCredit);
  PCollection<KV<Long, MusicBrainzDataObject>> recordingsJoinedWithCredits =
      MusicBrainzTransforms.by("artist_credit_name_artist", recordingsWithCredits);
  PCollection<MusicBrainzDataObject> artistsWithNestedRecordings = MusicBrainzTransforms.nest(artistsById, recordingsJoinedWithCredits, "recordings");


  DirectPipelineRunner.EvaluationResults results = p.run();

  List<MusicBrainzDataObject> resultObjects = results.getPCollection(artistsWithNestedRecordings);
  assert (resultObjects.size() == 1);
  assert (((List<MusicBrainzDataObject>) resultObjects.get(0).getColumnValue("artist_recordings")).size() == 448);


}
 

import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
private static Class<? extends PipelineRunner<?>> runner(String name) {
  Class<? extends PipelineRunner<?>> c = DirectPipelineRunner.class; // default

  if (DEFAULT_RUNNER.equals(name) || name == null) {
    c = DataflowPipelineRunner.class;
  } else if (BLOCKING_RUNNER.equals(name)) {
    c = BlockingDataflowPipelineRunner.class;
  } else if (DIRECT_RUNNER.equals(name)) {
    c = DirectPipelineRunner.class;
  }
  return c;
}
 

import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner; //导入依赖的package包/类
@Before public void init() throws IOException {
  sparkRunner = SparkPipelineRunner.create();
  directRunner = DirectPipelineRunner.createForTest();
  testDataDirName = Joiner.on(File.separator).join("target", "test-data", name.getMethodName())
      + File.separator;
  FileUtils.deleteDirectory(new File(testDataDirName));
  new File(testDataDirName).mkdirs();
}