import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * Returns the table to write, or {@code null} if writing with {@code tableFunction}.
 *
 * <p>If the table's project is not specified, use the executing project.
 */
@Nullable
ValueProvider<TableReference> getTableWithDefaultProject(BigQueryOptions bqOptions) {
  ValueProvider<TableReference> table = getTable();
  if (table == null) {
    return table;
  }

  if (!table.isAccessible()) {
    LOG.info("Using a dynamic value for table input. This must contain a project"
            + " in the table reference: {}", table);
    return table;
  }
  if (Strings.isNullOrEmpty(table.get().getProjectId())) {
    // If user does not specify a project we assume the table to be located in
    // the default project.
    TableReference tableRef = table.get();
    tableRef.setProjectId(bqOptions.getProject());
    return NestedValueProvider.of(StaticValueProvider.of(
        BigQueryHelpers.toJsonString(tableRef)), new JsonTableRefToTableRef());
  }
  return table;
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
BatchLoads(WriteDisposition writeDisposition, CreateDisposition createDisposition,
           boolean singletonTable,
           DynamicDestinations<?, DestinationT> dynamicDestinations,
           Coder<DestinationT> destinationCoder,
           ValueProvider<String> customGcsTempLocation) {
  bigQueryServices = new BigQueryServicesImpl();
  this.writeDisposition = writeDisposition;
  this.createDisposition = createDisposition;
  this.singletonTable = singletonTable;
  this.dynamicDestinations = dynamicDestinations;
  this.destinationCoder = destinationCoder;
  this.maxNumWritersPerBundle = DEFAULT_MAX_NUM_WRITERS_PER_BUNDLE;
  this.maxFileSize = DEFAULT_MAX_FILE_SIZE;
  this.numFileShards = DEFAULT_NUM_FILE_SHARDS;
  this.triggeringFrequency = null;
  this.customGcsTempLocation = customGcsTempLocation;
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * This function creates the DAG graph of transforms. It can be called from main()
 * as well as from the ControlPipeline.
 * @param options
 * @return
 * @throws Exception
 */
public static Pipeline createIndexerPipeline(FileIndexerPipelineOptions options) throws Exception {
	
    IndexerPipelineUtils.validateIndexerPipelineOptions(options);
	Pipeline pipeline = Pipeline.create(options);
	
	// PHASE: Read raw content from sources
	
	PCollection<InputContent> readContent = pipeline
			.apply("Read entire CSV file", org.apache.beam.sdk.io.Read.from(new RecordFileSource<String>(
				ValueProvider.StaticValueProvider.of(options.getInputFile()), 
				StringUtf8Coder.of(), RecordFileSource.DEFAULT_RECORD_SEPARATOR))) //
			.apply("Parse CSV file into InputContent objects", ParDo.of(new ParseCSVFile()));
	
	// Define the accumulators of all filters
	PCollection<InputContent> contentToIndex = readContent;
	
	// PHASE: Index documents (extract opinions and entities/tags). 
	// Return successfully indexed docs, and create a Bigtable write transform to store errors 
	// in Dead Letter table.
	PCollection<ContentIndexSummary> indexes = indexDocuments(options, contentToIndex);
	
	if (options.getRatioEnrichWithCNLP() > 0)
		indexes = enrichWithCNLP(indexes, options.getRatioEnrichWithCNLP());
	
	// PHASE: Write to BigQuery
	// For the Indexes that are unique ("filteredIndexes"), create records in webresource, document, and sentiment.
	// Then, merge resulting webresources with webresourceRowsUnindexed and webresourceDeduped
	indexes
		.apply(ParDo.of(new CreateCSVLineFromIndexSummaryFn()))
		.apply(TextIO.write()
			.to(options.getOutputFile()));
	
	
	return pipeline;
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@Override
public FileBasedSink.FilenamePolicy getFilenamePolicy(String genus) {
  return DefaultFilenamePolicy.fromStandardParameters(
    ValueProvider.StaticValueProvider.of(
      baseDir.resolve(genus, RESOLVE_FILE)),
    ShardNameTemplate.INDEX_OF_MAX,
    ".avro",
    false);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
private BigQueryQuerySource(
    String stepUuid,
    ValueProvider<String> query,
    Boolean flattenResults,
    Boolean useLegacySql,
    BigQueryServices bqServices,
    Coder<T> coder,
    SerializableFunction<SchemaAndRecord, T> parseFn) {
  super(stepUuid, bqServices, coder, parseFn);
  this.query = checkNotNull(query, "query");
  this.flattenResults = checkNotNull(flattenResults, "flattenResults");
  this.useLegacySql = checkNotNull(useLegacySql, "useLegacySql");
  this.dryRunJobStats = new AtomicReference<>();
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * Returns a new {@link ValueProvider} that is inaccessible before {@link #run}, but will be
 * accessible while the pipeline runs.
 */
public <T> ValueProvider<T> newProvider(T runtimeValue) {
  String uuid = UUID.randomUUID().toString();
  providerRuntimeValues.put(uuid, runtimeValue);
  return ValueProvider.NestedValueProvider.of(
      options.as(TestValueProviderOptions.class).getProviderRuntimeValues(),
      new GetFromRuntimeValues<T>(uuid));
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@VisibleForTesting
TFRecordSink(
    ValueProvider<ResourceId> outputPrefix,
    @Nullable String shardTemplate,
    @Nullable String suffix,
    Compression compression) {
  super(
      outputPrefix,
      DynamicFileDestinations.<byte[]>constant(
          DefaultFilenamePolicy.fromStandardParameters(
              outputPrefix, shardTemplate, suffix, false)),
      compression);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * Construct a {@link FileBasedSink} with the given temp directory, producing uncompressed files.
 */
@Experimental(Kind.FILESYSTEM)
public FileBasedSink(
    ValueProvider<ResourceId> tempDirectoryProvider,
    DynamicDestinations<?, DestinationT, OutputT> dynamicDestinations) {
  this(tempDirectoryProvider, dynamicDestinations, Compression.UNCOMPRESSED);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
XmlSource(
    ValueProvider<String> spec,
    XmlIO.MappingConfiguration<T> configuration,
    long minBundleSizeBytes) {
  super(spec, minBundleSizeBytes);
  this.configuration = configuration;
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@Test
public void testRuntimeOptionsNotCalledInApplyOutput() {
  p.enableAbandonedNodeEnforcement(false);

  BigQueryIO.Write<TableRow> write = BigQueryIO.writeTableRows()
      .to(p.newProvider("some-table"))
      .withSchema(ValueProvider.NestedValueProvider.of(
          p.newProvider("some-schema"), new BigQueryHelpers.JsonSchemaToTableSchema()))
      .withoutValidation();
  p.apply(Create.empty(TableRowJsonCoder.of())).apply(write);
  // Test that this doesn't throw.
  DisplayData.from(write);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/** Returns an IO transform that writes detail reports to registrar-tld keyed CSV files. */
private TextIO.TypedWrite<BillingEvent, Params> writeDetailReports(
    ValueProvider<String> yearMonthProvider) {
  return TextIO.<BillingEvent>writeCustomType()
      // TODO(larryruili): Replace with billing bucket/yyyy-MM after verifying 2017-12 output.
      .to(
          InvoicingUtils.makeDestinationFunction(beamBucket + "/results", yearMonthProvider),
          InvoicingUtils.makeEmptyDestinationParams(beamBucket + "/results"))
      .withFormatFunction(BillingEvent::toCsv)
      .withoutSharding()
      .withTempDirectory(FileBasedSink.convertToFileResourceIfPossible(beamBucket + "/temporary"))
      .withHeader(BillingEvent.getHeader())
      .withSuffix(".csv");
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@Test
public void testRuntimeValueProviderTopic() {
  TestPipeline pipeline = TestPipeline.create();
  ValueProvider<String> topic = pipeline.newProvider("projects/project/topics/topic");
  Read<String> pubsubRead = PubsubIO.readStrings().fromTopic(topic);
  pipeline.apply(pubsubRead);
  assertThat(pubsubRead.getTopicProvider(), not(nullValue()));
  assertThat(pubsubRead.getTopicProvider().isAccessible(), is(false));
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@Override
public void populateDisplayData(DisplayData.Builder builder) {
  super.populateDisplayData(builder);
  builder.addIfNotNull(DisplayData.item("topic", getTopic()).withLabel("Topic"));
  Set<String> ignoredProducerPropertiesKeys = IGNORED_PRODUCER_PROPERTIES.keySet();
  for (Map.Entry<String, Object> conf : getProducerConfig().entrySet()) {
    String key = conf.getKey();
    if (!ignoredProducerPropertiesKeys.contains(key)) {
      Object value = DisplayData.inferType(conf.getValue()) != null
          ? conf.getValue() : String.valueOf(conf.getValue());
      builder.add(DisplayData.item(key, ValueProvider.StaticValueProvider.of(value)));
    }
  }
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/** Like {@link #to(String)}. */
public TypedWrite<UserT, DestinationT, OutputT> to(ValueProvider<String> outputPrefix) {
  return toResource(
      NestedValueProvider.of(
          outputPrefix,
          // The function cannot be created as an anonymous class here since the enclosed class
          // may contain unserializable members.
          new OutputPrefixToResourceId()));
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
@Override
public WriteFilesResult<DestinationT> expand(PCollection<UserT> input) {
  checkArgument(
      getFilenamePrefix() != null || getTempDirectory() != null,
      "Need to set either the filename prefix or the tempDirectory of a AvroIO.Write "
          + "transform.");
  if (getFilenamePolicy() != null) {
    checkArgument(
        getShardTemplate() == null && getFilenameSuffix() == null,
        "shardTemplate and filenameSuffix should only be used with the default "
            + "filename policy");
  }
  if (getDynamicDestinations() != null) {
    checkArgument(
        getFormatFunction() == null,
        "A format function should not be specified "
            + "with DynamicDestinations. Use DynamicDestinations.formatRecord instead");
  } else {
    checkArgument(
        getSchema() != null, "Unless using DynamicDestinations, .withSchema() is required.");
  }

  ValueProvider<ResourceId> tempDirectory = getTempDirectory();
  if (tempDirectory == null) {
    tempDirectory = getFilenamePrefix();
  }
  WriteFiles<UserT, DestinationT, OutputT> write =
      WriteFiles.to(
          new AvroSink<>(tempDirectory, resolveDynamicDestinations(), getGenericRecords()));
  if (getNumShards() > 0) {
    write = write.withNumShards(getNumShards());
  }
  if (getWindowedWrites()) {
    write = write.withWindowedWrites();
  }
  return input.apply("Write", write);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * Like {@code topic()} but with a {@link ValueProvider}.
 */
public Read<T> fromTopic(ValueProvider<String> topic) {
  if (topic.isAccessible()) {
    // Validate.
    PubsubTopic.fromPath(topic.get());
  }
  return toBuilder()
      .setTopicProvider(NestedValueProvider.of(topic, new TopicTranslator()))
      .build();
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
WriterFn(
    PubsubClientFactory pubsubFactory,
    ValueProvider<TopicPath> topic,
    String timestampAttribute,
    String idAttribute,
    int publishBatchSize,
    int publishBatchBytes) {
  this.pubsubFactory = pubsubFactory;
  this.topic = topic;
  this.timestampAttribute = timestampAttribute;
  this.idAttribute = idAttribute;
  this.publishBatchSize = publishBatchSize;
  this.publishBatchBytes = publishBatchBytes;
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
public PubsubUnboundedSink(
    PubsubClientFactory pubsubFactory,
    ValueProvider<TopicPath> topic,
    String timestampAttribute,
    String idAttribute,
    int numShards) {
  this(pubsubFactory, topic, timestampAttribute, idAttribute, numShards,
       DEFAULT_PUBLISH_BATCH_SIZE, DEFAULT_PUBLISH_BATCH_BYTES, DEFAULT_MAX_LATENCY,
       RecordIdMethod.RANDOM);
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
/**
 * Returns a function mapping from {@code BillingEvent} to filename {@code Params}.
 *
 * <p>Beam uses this to determine which file a given {@code BillingEvent} should get placed into.
 *
 * @param outputBucket the GCS bucket we're outputting reports to
 * @param yearMonthProvider a runtime provider for the yyyy-MM we're generating the invoice for
 */
static SerializableFunction<BillingEvent, Params> makeDestinationFunction(
    String outputBucket, ValueProvider<String> yearMonthProvider) {
  return billingEvent ->
      new Params()
          .withShardTemplate("")
          .withSuffix(".csv")
          .withBaseFilename(
              NestedValueProvider.of(
                  yearMonthProvider,
                  yearMonth ->
                      FileBasedSink.convertToFileResourceIfPossible(
                          String.format(
                              "%s/%s", outputBucket, billingEvent.toFilename(yearMonth)))));
}
 

import org.apache.beam.sdk.options.ValueProvider; //导入依赖的package包/类
private ViaFileBasedSink(
    Write<DestinationT, UserT> spec) {
  super(
      ValueProvider.NestedValueProvider.of(
          spec.getTempDirectory(),
          new SerializableFunction<String, ResourceId>() {
            @Override
            public ResourceId apply(String input) {
              return FileSystems.matchNewResource(input, true /* isDirectory */);
            }
          }),
      new DynamicDestinationsAdapter<UserT, DestinationT, OutputT>(spec),
      spec.getCompression());
  this.spec = spec;
}