import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
public static void main(String[] args) {
  Pipeline p = Pipeline.create(
      PipelineOptionsFactory.fromArgs(args).withValidation().create());

  p.apply(Create.of("Hello", "World"))
  .apply(MapElements.via(new SimpleFunction<String, String>() {
    @Override
    public String apply(String input) {
      return input.toUpperCase();
    }
  }))
  .apply(ParDo.of(new DoFn<String, Void>() {
    @ProcessElement
    public void processElement(ProcessContext c)  {
      LOG.info(c.element());
    }
  }));

  p.run();
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
@Override
public PCollection<T> expand(PCollection<T> input) {
  WindowingStrategy<?, ?> outputWindowingStrategy = getOutputWindowing(
      input.getWindowingStrategy());

  return input
      // We first apply a (trivial) transform to the input PCollection to produce a new
      // PCollection. This ensures that we don't modify the windowing strategy of the input
      // which may be used elsewhere.
      .apply("Identity", MapElements.via(new SimpleFunction<T, T>() {
        @Override public T apply(T element) {
          return element;
        }
      }))
      // Then we modify the windowing strategy.
      .setWindowingStrategyInternal(outputWindowingStrategy);
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
public PCollectionSingletonIterableAssert(
    PCollection<Iterable<T>> actual,
    AssertionWindows rewindowingStrategy,
    SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
        paneExtractor,
    PAssertionSite site) {
  this.actual = actual;

  @SuppressWarnings("unchecked")
  Coder<T> typedCoder = (Coder<T>) actual.getCoder().getCoderArguments().get(0);
  this.elementCoder = typedCoder;

  this.rewindowingStrategy = rewindowingStrategy;
  this.paneExtractor = paneExtractor;
  this.site = site;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
@BeforeClass
public static void setUp() throws IOException, InterruptedException {
  serConf = loadTestConfiguration(
                EmployeeInputFormat.class,
                Text.class,
                Employee.class);
  myKeyTranslate = new SimpleFunction<Text, String>() {
    @Override
    public String apply(Text input) {
      return input.toString();
    }
  };
  myValueTranslate = new SimpleFunction<Employee, String>() {
    @Override
    public String apply(Employee input) {
      return input.getEmpName() + "_" + input.getEmpAddress();
    }
  };
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
/**
 * This test validates functionality of {@link HadoopInputFormatIO.Read#validateTransform()
 * Read.validateTransform()} function when myKeyTranslate's (simple function provided by user for
 * key translation) input type is not same as Hadoop InputFormat's keyClass(Which is property set
 * in configuration as "key.class").
 */
@Test
public void testReadValidationFailsWithWrongInputTypeKeyTranslationFunction() {
  SimpleFunction<LongWritable, String> myKeyTranslateWithWrongInputType =
      new SimpleFunction<LongWritable, String>() {
        @Override
        public String apply(LongWritable input) {
          return input.toString();
        }
      };
  HadoopInputFormatIO.Read<String, Employee> read = HadoopInputFormatIO.<String, Employee>read()
      .withConfiguration(serConf.get())
      .withKeyTranslation(myKeyTranslateWithWrongInputType);
  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage(String.format(
      "Key translation's input type is not same as hadoop InputFormat : %s key " + "class : %s",
      serConf.get().getClass("mapreduce.job.inputformat.class",
          InputFormat.class), serConf.get()
          .getClass("key.class", Object.class)));
  read.validateTransform();
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
/**
 * This test validates functionality of {@link HadoopInputFormatIO.Read#validateTransform()
 * Read.validateTransform()} function when myValueTranslate's (simple function provided by user
 * for value translation) input type is not same as Hadoop InputFormat's valueClass(Which is
 * property set in configuration as "value.class").
 */
@Test
public void testReadValidationFailsWithWrongInputTypeValueTranslationFunction() {
  SimpleFunction<LongWritable, String> myValueTranslateWithWrongInputType =
      new SimpleFunction<LongWritable, String>() {
        @Override
        public String apply(LongWritable input) {
          return input.toString();
        }
      };
  HadoopInputFormatIO.Read<Text, String> read =
      HadoopInputFormatIO.<Text, String>read()
          .withConfiguration(serConf.get())
          .withValueTranslation(myValueTranslateWithWrongInputType);
  String expectedMessage =
      String.format(
          "Value translation's input type is not same as hadoop InputFormat :  "
              + "%s value class : %s",
          serConf.get().getClass("mapreduce.job.inputformat.class",
              InputFormat.class),
          serConf.get().getClass("value.class", Object.class));
  thrown.expect(IllegalArgumentException.class);
  thrown.expectMessage(expectedMessage);
  read.validateTransform();
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
@Test
public void testCreateNeverWithStreaming() throws Exception {
  p.enableAbandonedNodeEnforcement(false);

  TableReference tableRef = new TableReference();
  tableRef.setDatasetId("dataset");
  tableRef.setTableId("sometable");

  PCollection<TableRow> tableRows =
      p.apply(GenerateSequence.from(0))
          .apply(
              MapElements.via(
                  new SimpleFunction<Long, TableRow>() {
                    @Override
                    public TableRow apply(Long input) {
                      return null;
                    }
                  }))
          .setCoder(TableRowJsonCoder.of());
  tableRows
      .apply(BigQueryIO.writeTableRows().to(tableRef)
          .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
          .withoutValidation());
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
public PCollectionContentsAssert(
    PCollection<T> actual,
    AssertionWindows rewindowingStrategy,
    SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor,
    PAssertionSite site) {
  this.actual = actual;
  this.rewindowingStrategy = rewindowingStrategy;
  this.paneExtractor = paneExtractor;
  this.site = site;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private PCollectionContentsAssert<T> withPane(
    BoundedWindow window,
    SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor) {
  @SuppressWarnings({"unchecked", "rawtypes"})
  Coder<BoundedWindow> windowCoder =
      (Coder) actual.getWindowingStrategy().getWindowFn().windowCoder();
  return new PCollectionContentsAssert<>(
      actual, IntoStaticWindows.of(windowCoder, window), paneExtractor, site);
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private PCollectionSingletonIterableAssert<T> withPanes(
    BoundedWindow window,
    SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
        paneExtractor) {
  @SuppressWarnings({"unchecked", "rawtypes"})
  Coder<BoundedWindow> windowCoder =
      (Coder) actual.getWindowingStrategy().getWindowFn().windowCoder();
  return new PCollectionSingletonIterableAssert<>(
      actual, IntoStaticWindows.of(windowCoder, window), paneExtractor, site);
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private PCollectionViewAssert(
    PCollection<ElemT> actual,
    PTransform<PCollection<ElemT>, PCollectionView<ViewT>> view,
    AssertionWindows rewindowActuals,
    SimpleFunction<Iterable<ValueInSingleWindow<ElemT>>, Iterable<ElemT>> paneExtractor,
    Coder<ViewT> coder,
    PAssertionSite site) {
  this.actual = actual;
  this.view = view;
  this.rewindowActuals = rewindowActuals;
  this.paneExtractor = paneExtractor;
  this.coder = coder;
  this.site = site;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private PCollectionViewAssert<ElemT, ViewT> inPane(
    BoundedWindow window,
    SimpleFunction<Iterable<ValueInSingleWindow<ElemT>>, Iterable<ElemT>> paneExtractor) {
  return new PCollectionViewAssert<>(
      actual,
      view,
      IntoStaticWindows.of(
          (Coder) actual.getWindowingStrategy().getWindowFn().windowCoder(), window),
      paneExtractor,
      coder,
      site);
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
public static <T, ActualT> CreateActual<T, ActualT> from(
    PCollection<T> actual,
    AssertionWindows rewindowActuals,
    SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> extractPane,
    PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
  return new CreateActual<>(actual, rewindowActuals, extractPane, actualView);
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private CreateActual(
    PCollection<T> actual,
    AssertionWindows rewindowActuals,
    SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> extractPane,
    PTransform<PCollection<T>, PCollectionView<ActualT>> actualView) {
  this.actual = actual;
  this.rewindowActuals = rewindowActuals;
  this.extractPane = extractPane;
  this.actualView = actualView;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private GroupThenAssert(
    SerializableFunction<Iterable<T>, Void> checkerFn,
    AssertionWindows rewindowingStrategy,
    SimpleFunction<Iterable<ValueInSingleWindow<T>>, Iterable<T>> paneExtractor,
    PAssertionSite site) {
  this.checkerFn = checkerFn;
  this.rewindowingStrategy = rewindowingStrategy;
  this.paneExtractor = paneExtractor;
  this.site = site;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private GroupThenAssertForSingleton(
    SerializableFunction<Iterable<T>, Void> checkerFn,
    AssertionWindows rewindowingStrategy,
    SimpleFunction<Iterable<ValueInSingleWindow<Iterable<T>>>, Iterable<Iterable<T>>>
        paneExtractor,
    PAssertionSite site) {
  this.checkerFn = checkerFn;
  this.rewindowingStrategy = rewindowingStrategy;
  this.paneExtractor = paneExtractor;
  this.site = site;
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
@Test
public void testExpandHasMatchingTags() {
  TupleTag<Integer> intTag = new TupleTag<>();
  TupleTag<String> strTag = new TupleTag<>();
  TupleTag<Long> longTag = new TupleTag<>();

  Pipeline p = TestPipeline.create();
  PCollection<Long> longs = p.apply(GenerateSequence.from(0).to(100));
  PCollection<String> strs = p.apply(Create.of("foo", "bar", "baz"));
  PCollection<Integer> ints = longs.apply(MapElements.via(new SimpleFunction<Long, Integer>() {
    @Override
    public Integer apply(Long input) {
      return input.intValue();
    }
  }));

  Map<TupleTag<?>, PCollection<?>> pcsByTag =
      ImmutableMap.<TupleTag<?>, PCollection<?>>builder()
          .put(strTag, strs)
          .put(intTag, ints)
          .put(longTag, longs)
          .build();
  PCollectionTuple tuple =
      PCollectionTuple.of(intTag, ints).and(longTag, longs).and(strTag, strs);
  assertThat(tuple.getAll(), equalTo(pcsByTag));
  PCollectionTuple reconstructed = PCollectionTuple.empty(p);
  for (Entry<TupleTag<?>, PValue> taggedValue : tuple.expand().entrySet()) {
    TupleTag<?> tag = taggedValue.getKey();
    PValue value = taggedValue.getValue();
    assertThat("The tag should map back to the value", tuple.get(tag), equalTo(value));
    assertThat(value, Matchers.<PValue>equalTo(pcsByTag.get(tag)));
    reconstructed = reconstructed.and(tag, (PCollection) value);
  }

  assertThat(reconstructed, equalTo(tuple));
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private static PTransform<PCollection<? extends String>, PCollection<String>> addSuffix(
    final String suffix) {
  return MapElements.via(new SimpleFunction<String, String>() {
    @Override
    public String apply(String input) {
      return input + suffix;
    }
  });
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
@SuppressWarnings("UnusedReturnValue")
private static PCollection<String> addTransform(final PCollection<String> pCollection) {
  return pCollection.apply(
      "Map2",
      MapElements.via(
          new SimpleFunction<String, String>() {

            @Override
            public String apply(final String input) {
              return WHATEVER;
            }
          }));
}
 

import org.apache.beam.sdk.transforms.SimpleFunction; //导入依赖的package包/类
private static PCollection<String> pCollection(final Pipeline pipeline) {
  return pipeline
      .apply("Create", Create.of(WORDS).withCoder(StringUtf8Coder.of()))
      .apply(
          "Map1",
          MapElements.via(
              new SimpleFunction<String, String>() {

                @Override
                public String apply(final String input) {
                  return WHATEVER;
                }
              }));
}