import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
/**
 * Creates and returns a new SamzaRunner with provided options.
 *
 * @param options The PipelineOptions to use when executing the job.
 * @return A pipeline runner that will execute with specified options.
 */
public static SamzaRunner fromOptions(PipelineOptions options) {
  SamzaPipelineOptions samzaOptions = PipelineOptionsValidator.validate(SamzaPipelineOptions.class, options);

  ArrayList<String> missing = new ArrayList<>();

  if (samzaOptions.getAppName() == null) {
    missing.add("appName");
  }
  if (missing.size() > 0) {
    throw new IllegalArgumentException(
        "Missing required values: " + Joiner.on(',').join(missing));
  }

  return new SamzaRunner(samzaOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
/**
 * Constructs a runner from the provided {@link PipelineOptions}.
 *
 * @return The newly created runner.
 */
public static PipelineRunner<? extends PipelineResult> fromOptions(PipelineOptions options) {
  checkNotNull(options);
  PipelineOptionsValidator.validate(PipelineOptions.class, options);

  // (Re-)register standard FileSystems. Clobbers any prior credentials.
  FileSystems.setDefaultPipelineOptions(options);

  @SuppressWarnings("unchecked")
  PipelineRunner<? extends PipelineResult> result =
      InstanceBuilder.ofType(PipelineRunner.class)
      .fromClass(options.getRunner())
      .fromFactoryMethod("fromOptions")
      .withArg(PipelineOptions.class, options)
      .build();
  return result;
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
/**
 * Creates and returns a new SparkRunner with specified options.
 *
 * @param options The PipelineOptions to use when executing the job.
 * @return A pipeline runner that will execute with specified options.
 */
public static SparkRunner fromOptions(PipelineOptions options) {
  SparkPipelineOptions sparkOptions =
      PipelineOptionsValidator.validate(SparkPipelineOptions.class, options);

  if (sparkOptions.getFilesToStage() == null) {
    sparkOptions.setFilesToStage(detectClassPathResourcesToStage(
        SparkRunner.class.getClassLoader()));
    LOG.info("PipelineOptions.filesToStage was not specified. "
            + "Defaulting to files from the classpath: will stage {} files. "
            + "Enable logging at DEBUG level to see which files will be staged.",
        sparkOptions.getFilesToStage().size());
    LOG.debug("Classpath elements: {}", sparkOptions.getFilesToStage());
  }

  return new SparkRunner(sparkOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static IOTestPipelineOptions readTestPipelineOptions() {
  PipelineOptionsFactory.register(IOTestPipelineOptions.class);
  IOTestPipelineOptions options = TestPipeline
      .testingPipelineOptions()
      .as(IOTestPipelineOptions.class);

  return PipelineOptionsValidator.validate(IOTestPipelineOptions.class, options);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static SparkRunnerDebugger fromOptions(PipelineOptions options) {
  if (options instanceof TestSparkPipelineOptions) {
    TestSparkPipelineOptions testSparkPipelineOptions =
        PipelineOptionsValidator.validate(TestSparkPipelineOptions.class, options);
    return new SparkRunnerDebugger(testSparkPipelineOptions);
  } else {
    SparkPipelineOptions sparkPipelineOptions =
        PipelineOptionsValidator.validate(SparkPipelineOptions.class, options);
    return new SparkRunnerDebugger(sparkPipelineOptions);
  }
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
/**
 * Construct a runner from the provided options.
 *
 * @param options Properties which configure the runner.
 * @return The newly created runner.
 */
public static FlinkRunner fromOptions(PipelineOptions options) {
  FlinkPipelineOptions flinkOptions =
      PipelineOptionsValidator.validate(FlinkPipelineOptions.class, options);
  ArrayList<String> missing = new ArrayList<>();

  if (flinkOptions.getAppName() == null) {
    missing.add("appName");
  }
  if (missing.size() > 0) {
    throw new IllegalArgumentException(
        "Missing required values: " + Joiner.on(',').join(missing));
  }

  if (flinkOptions.getFilesToStage() == null) {
    flinkOptions.setFilesToStage(detectClassPathResourcesToStage(
        FlinkRunner.class.getClassLoader()));
    LOG.info("PipelineOptions.filesToStage was not specified. "
            + "Defaulting to files from the classpath: will stage {} files. "
            + "Enable logging at DEBUG level to see which files will be staged.",
        flinkOptions.getFilesToStage().size());
    LOG.debug("Classpath elements: {}", flinkOptions.getFilesToStage());
  }

  // Set Flink Master to [auto] if no option was specified.
  if (flinkOptions.getFlinkMaster() == null) {
    flinkOptions.setFlinkMaster("[auto]");
  }

  return new FlinkRunner(flinkOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static ApexRunner fromOptions(PipelineOptions options) {
  ApexPipelineOptions apexPipelineOptions =
          PipelineOptionsValidator.validate(ApexPipelineOptions.class, options);
  return new ApexRunner(apexPipelineOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static TestApexRunner fromOptions(PipelineOptions options) {
  ApexPipelineOptions apexOptions = PipelineOptionsValidator
      .validate(ApexPipelineOptions.class, options);
  return new TestApexRunner(apexOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static TestFlinkRunner fromOptions(PipelineOptions options) {
  FlinkPipelineOptions flinkOptions =
      PipelineOptionsValidator.validate(FlinkPipelineOptions.class, options);
  return new TestFlinkRunner(flinkOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static GearpumpRunner fromOptions(PipelineOptions options) {
  GearpumpPipelineOptions pipelineOptions =
      PipelineOptionsValidator.validate(GearpumpPipelineOptions.class, options);
  return new GearpumpRunner(pipelineOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
public static TestGearpumpRunner fromOptions(PipelineOptions options) {
  GearpumpPipelineOptions pipelineOptions =
      PipelineOptionsValidator.validate(GearpumpPipelineOptions.class, options);
  return new TestGearpumpRunner(pipelineOptions);
}
 

import org.apache.beam.sdk.options.PipelineOptionsValidator; //导入依赖的package包/类
@Override
  public void populateDAG(DAG dag, Configuration conf)
  {
/*
    HadoopFileSystemOptions fsoptions = PipelineOptionsFactory.as(HadoopFileSystemOptions.class);
    //fsoptions.setHdfsConfiguration(ImmutableList.of(new Configuration()));
    for (FileSystemRegistrar registrar
        : Lists.newArrayList(ServiceLoader.load(FileSystemRegistrar.class).iterator())) {
      System.out.println(registrar);
      if (registrar instanceof HadoopFileSystemRegistrar) {
        Iterable<FileSystem> fileSystems = registrar.fromOptions(fsoptions);
        for (FileSystem fs : fileSystems) {
          System.out.println(fs);
        }
      }
    }
*/
    String optionsStr = conf.get(KEY_PIPELINE_OPTIONS, "--runner=ApexRunner");
    String[] args = StringUtils.splitByWholeSeparator(optionsStr, " ");
    WordCountOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
        .as(WordCountOptions.class);
    Pipeline p = Pipeline.create(options);
    // Concepts #2 and #3: Our pipeline applies the composite CountWords transform, and passes the
    // static FormatAsTextFn() to the ParDo transform.
      p.apply("ReadFromHDFS", TextIO.read().from(absoluteUri(options.getInputFile())))
      .apply(new CountWords())
      .apply(MapElements.via(new FormatAsTextFn()))
      .apply("WriteToHDFS", TextIO.write().to(absoluteUri(options.getOutput())).withNumShards(2));

    ApexPipelineOptions apexPipelineOptions =
        PipelineOptionsValidator.validate(ApexPipelineOptions.class, options);
    final ApexPipelineTranslator translator = new ApexPipelineTranslator(apexPipelineOptions);

    // roundabout way to apply overrides - we just want to translate, not run the pipeline here
    ApexRunner runner = new ApexRunner(apexPipelineOptions);
    try {
      Method m = ApexRunner.class.getDeclaredMethod("getOverrides");
      m.setAccessible(true);
      @SuppressWarnings("unchecked")
      List<PTransformOverride> overrides = (List<PTransformOverride>)m.invoke(runner);
      p.replaceAll(overrides);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    translator.translate(p, dag);
  }