import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<InputT> teamAndScore) {
  teamAndScore
      .apply("ConvertToRow", ParDo.of(new BuildRowFn()))
      .apply(
          BigQueryIO.writeTableRows()
              .to(getTable(projectId, datasetId, tableName))
              .withSchema(getSchema())
              .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
              .withWriteDisposition(WriteDisposition.WRITE_APPEND));
  return PDone.in(teamAndScore.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<KV<String, KV<URI, Double>>> wordToUriAndTfIdf) {
  return wordToUriAndTfIdf
      .apply("Format", ParDo.of(new DoFn<KV<String, KV<URI, Double>>, String>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
          c.output(String.format("%s,\t%s,\t%f",
              c.element().getKey(),
              c.element().getValue().getKey(),
              c.element().getValue().getValue()));
        }
      }))
      .apply(TextIO.write()
          .to(output)
          .withSuffix(".csv"));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PTransform<? super PCollection<BeamRecord>, PDone> buildIOWriter() {
  checkArgument(topics != null && topics.size() == 1,
      "Only one topic can be acceptable as output.");

  return new PTransform<PCollection<BeamRecord>, PDone>() {
    @Override
    public PDone expand(PCollection<BeamRecord> input) {
      return input.apply("out_reformat", getPTransformForOutput()).apply("persistent",
          KafkaIO.<byte[], byte[]>write()
              .withBootstrapServers(bootstrapServers)
              .withTopic(topics.get(0))
              .withKeySerializer(ByteArraySerializer.class)
              .withValueSerializer(ByteArraySerializer.class));
    }
  };
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<BeamRecord> input) {
  input.apply(ParDo.of(new DoFn<BeamRecord, Void>() {
    @ProcessElement
    public void processElement(ProcessContext c) {
      CONTENT.add(c.element());
    }

    @Teardown
    public void close() {
      CONTENT.clear();
    }

  }));
  return PDone.in(input.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<byte[]> input) {
  checkState(getOutputPrefix() != null,
      "need to set the output prefix of a TFRecordIO.Write transform");
  WriteFiles<byte[], Void, byte[]> write =
      WriteFiles.to(
          new TFRecordSink(
              getOutputPrefix(),
              getShardTemplate(),
              getFilenameSuffix(),
              getCompression()));
  if (getNumShards() > 0) {
    write = write.withNumShards(getNumShards());
  }
  input.apply("Write", write);
  return PDone.in(input.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Test
public void testPTransformNameWithAnonOuterClass() throws Exception {
  AnonymousClass anonymousClassObj = new AnonymousClass() {
    class NamedInnerClass extends PTransform<PBegin, PDone> {
      @Override
      public PDone expand(PBegin begin) {
        throw new IllegalArgumentException("Should never be applied");
      }
    }

    @Override
    public Object getInnerClassInstance() {
      return new NamedInnerClass();
    }
  };

  assertEquals("NamedInnerClass",
      NameUtils.approximateSimpleName(anonymousClassObj.getInnerClassInstance()));
  assertEquals("NameUtilsTest.NamedInnerClass",
      NameUtils.approximatePTransformName(anonymousClassObj.getInnerClassInstance().getClass()));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<T> input) {
  if (getTopicProvider() == null) {
    throw new IllegalStateException("need to set the topic of a PubsubIO.Write transform");
  }
  switch (input.isBounded()) {
    case BOUNDED:
      input.apply(ParDo.of(new PubsubBoundedWriter()));
      return PDone.in(input.getPipeline());
    case UNBOUNDED:
      return input.apply(MapElements.via(getFormatFn())).apply(new PubsubUnboundedSink(
          FACTORY,
          NestedValueProvider.of(getTopicProvider(), new TopicPathTranslator()),
          getTimestampAttribute(),
          getIdAttribute(),
          100 /* numShards */));
  }
  throw new RuntimeException(); // cases are exhaustive.
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<T> input) {
  checkArgument(getRecordClass() != null, "withRecordClass() is required");
  checkArgument(getRootElement() != null, "withRootElement() is required");
  checkArgument(getFilenamePrefix() != null, "to() is required");
  checkArgument(getCharset() != null, "withCharset() is required");
  try {
    JAXBContext.newInstance(getRecordClass());
  } catch (JAXBException e) {
    throw new RuntimeException("Error binding classes to a JAXB Context.", e);
  }

  ResourceId prefix =
      FileSystems.matchNewResource(getFilenamePrefix(), false /* isDirectory */);
  input.apply(
      FileIO.<T>write()
          .via(
              sink(getRecordClass())
                  .withCharset(Charset.forName(getCharset()))
                  .withRootElement(getRootElement()))
          .to(prefix.getCurrentDirectory().toString())
          .withPrefix(prefix.getFilename())
          .withSuffix(".xml")
          .withIgnoreWindowing());
  return PDone.in(input.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<KV<K, V>> input) {
  checkArgument(
    getProducerConfig().get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG) != null,
    "withBootstrapServers() is required");
  checkArgument(getTopic() != null, "withTopic() is required");
  checkArgument(getKeySerializer() != null, "withKeySerializer() is required");
  checkArgument(getValueSerializer() != null, "withValueSerializer() is required");

  if (isEOS()) {
    EOSWrite.ensureEOSSupport();

    // TODO: Verify that the group_id does not have existing state stored on Kafka unless
    //       this is an upgrade. This avoids issues with simple mistake of reusing group_id
    //       across multiple runs or across multiple jobs. This is checked when the sink
    //       transform initializes while processing the output. It might be better to
    //       check here to catch common mistake.

    input.apply(new EOSWrite<>(this));
  } else {
    input.apply(ParDo.of(new KafkaWriter<>(this)));
  }
  return PDone.in(input.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
private static AppliedPTransform<?, ?, ?> rawPTransformWithNullSpec(Pipeline pipeline) {
  PTransformTranslation.RawPTransform<PBegin, PDone> rawPTransform =
      new PTransformTranslation.RawPTransform<PBegin, PDone>() {
        @Override
        public String getUrn() {
          return "fake/urn";
        }

        @Nullable
        @Override
        public RunnerApi.FunctionSpec getSpec() {
          return null;
        }
      };
  return AppliedPTransform.<PBegin, PDone, PTransform<PBegin, PDone>>of(
      "RawPTransformWithNoSpec",
      pipeline.begin().expand(),
      PDone.in(pipeline).expand(),
      rawPTransform,
      pipeline);
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<String> input) {
  // Verify that the input has a compatible window type.
  checkArgument(
      input.getWindowingStrategy().getWindowFn().windowCoder() == IntervalWindow.getCoder());

  // filenamePrefix may contain a directory and a filename component. Pull out only the filename
  // component from that path for the PerWindowFiles.
  String prefix = "";
  ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(filenamePrefix);
  if (!resource.isDirectory()) {
    prefix = verifyNotNull(
        resource.getFilename(),
        "A non-directory resource should have a non-null filename: %s",
        resource);
  }

  return input.apply(
      TextIO.write()
          .to(resource.getCurrentDirectory())
          .withFilenamePolicy(new PerWindowFiles(prefix))
          .withWindowedWrites()
          .withNumShards(3));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<String> input) {
  // filenamePrefix may contain a directory and a filename component. Pull out only the filename
  // component from that path for the PerWindowFiles.
  String prefix = "";
  ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(filenamePrefix);
  if (!resource.isDirectory()) {
    prefix = verifyNotNull(
        resource.getFilename(),
        "A non-directory resource should have a non-null filename: %s",
        resource);
  }


  TextIO.Write write = TextIO.write()
      .to(resource.getCurrentDirectory())
      .withFilenamePolicy(new PerWindowFiles(prefix))
      .withWindowedWrites();
  if (numShards != null) {
    write = write.withNumShards(numShards);
  }
  return input.apply(write);
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<IndexedRecord> input) {
    return input.apply(JdbcIO.<IndexedRecord> write().withDataSourceConfiguration(JdbcIO.DataSourceConfiguration
            .create(properties.getDatasetProperties().getDatastoreProperties().getCurrentDriverClass(),
                    properties.getDatasetProperties().getDatastoreProperties().jdbcUrl.getValue())
            .withUsername(properties.getDatasetProperties().getDatastoreProperties().userId.getValue())
            .withPassword(properties.getDatasetProperties().getDatastoreProperties().password.getValue()))
            .withStatement(
                    JDBCSQLBuilder.getInstance().generateSQL4Insert(properties.getDatasetProperties().tableName.getValue(),
                            properties.getDatasetProperties().main.schema.getValue()))
            .withPreparedStatementSetter(new PreparedStatementSetter<IndexedRecord>() {

                @Override
                public void setParameters(IndexedRecord indexedRecord, PreparedStatement preparedStatement) throws Exception {
                    List<Schema.Field> fields = indexedRecord.getSchema().getFields();
                    int index = 0;
                    for (Schema.Field f : fields) {
                        JDBCMapping.setValue(++index, preparedStatement, f, indexedRecord.get(f.pos()));
                    }
                }
            }));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<KV<KV<String, String>, Long>> similarPairs) {
  return similarPairs
      .apply(Sum.<KV<String, String>>longsPerKey())
      .apply(Combine.globally(TO_LIST))
      .apply("PCoAAnalysis", ParDo.of(new PCoAnalysis(dataIndices)))
      .apply("FormatGraphData", ParDo
          .of(new DoFn<Iterable<PCoAnalysis.GraphResult>, String>() {
            @ProcessElement
            public void processElement(ProcessContext c) throws Exception {
              Iterable<PCoAnalysis.GraphResult> graphResults = c.element();
              for (PCoAnalysis.GraphResult result : graphResults) {
                c.output(result.toString());
              }
            }
          }))
      .apply("WriteCounts", TextIO.write().to(outputFile));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<String> line) {

  return line
      .apply(ParDo.of(new ParseEventFn()))
      .apply(ParDo.of(new KeyScoreByTeamFn()))
      .apply(Sum.<String>integersPerKey())
      .apply(ToString.kvs())
      .apply(TextIO.write().to(filepath)
          .withWindowedWrites().withNumShards(3)
          .withFilenamePolicy(new PerWindowFiles("count")));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<T> teamAndScore) {
  teamAndScore
    .apply("ConvertToRow", ParDo.of(new BuildRowFn()))
    .apply(BigQueryIO.writeTableRows()
              .to(getTable(projectId, datasetId, tableName))
              .withSchema(getSchema())
              .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
              .withWriteDisposition(WriteDisposition.WRITE_APPEND));
  return PDone.in(teamAndScore.getPipeline());
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<String> input) {
  // Verify that the input has a compatible window type.
  checkArgument(
      input.getWindowingStrategy().getWindowFn().windowCoder() == IntervalWindow.getCoder());

  ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(filenamePrefix);

  return input.apply(
      TextIO.write()
          .to(new PerWindowFiles(resource))
          .withTempDirectory(resource.getCurrentDirectory())
          .withWindowedWrites()
          .withNumShards(3));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<String> input) {
  ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(filenamePrefix);
  TextIO.Write write =
      TextIO.write()
          .to(new PerWindowFiles(resource))
          .withTempDirectory(resource.getCurrentDirectory())
          .withWindowedWrites();
  if (numShards != null) {
    write = write.withNumShards(numShards);
  }
  return input.apply(write);
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override public PDone expand(PCollection<BeamRecord> input) {
  return input.apply("encodeRecord", ParDo.of(new DoFn<BeamRecord, String>() {

    @ProcessElement public void processElement(ProcessContext ctx) {
      BeamRecord row = ctx.element();
      ctx.output(beamRecord2CsvLine(row, csvFormat));
    }
  })).apply(TextIO.write().to(filePattern));
}
 

import org.apache.beam.sdk.values.PDone; //导入依赖的package包/类
@Override
public PDone expand(PCollection<T> input) {
  input
      .apply("GroupGlobally", new GroupGlobally<T>(rewindowingStrategy))
      .apply("GetPane", MapElements.via(paneExtractor))
      .setCoder(IterableCoder.of(input.getCoder()))
      .apply("RunChecks", ParDo.of(new GroupedValuesCheckerDoFn<>(checkerFn, site)))
      .apply("VerifyAssertions", new DefaultConcludeTransform());

  return PDone.in(input.getPipeline());
}