import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
@Category({NeedsRunner.class, UsesTestStream.class})
public void testProcessingTimeTrigger() {
  TestStream<Long> source = TestStream.create(VarLongCoder.of())
      .addElements(TimestampedValue.of(1L, new Instant(1000L)),
          TimestampedValue.of(2L, new Instant(2000L)))
      .advanceProcessingTime(Duration.standardMinutes(12))
      .addElements(TimestampedValue.of(3L, new Instant(3000L)))
      .advanceProcessingTime(Duration.standardMinutes(6))
      .advanceWatermarkToInfinity();

  PCollection<Long> sum = p.apply(source)
      .apply(Window.<Long>configure().triggering(AfterWatermark.pastEndOfWindow()
          .withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
              .plusDelayOf(Duration.standardMinutes(5)))).accumulatingFiredPanes()
          .withAllowedLateness(Duration.ZERO))
      .apply(Sum.longsGlobally());

  PAssert.that(sum).inEarlyGlobalWindowPanes().containsInAnyOrder(3L, 6L);

  p.run();
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void usesMatcher() {
  final AtomicBoolean matcherUsed = new AtomicBoolean();
  Matcher<Integer> matcher =
      new TypeSafeMatcher<Integer>() {
        @Override
        public void describeTo(Description description) {}

        @Override
        protected boolean matchesSafely(Integer item) {
          matcherUsed.set(true);
          return item == 30;
        }
      };
  CombineFnTester.testCombineFn(
      Sum.ofIntegers(), Arrays.asList(1, 1, 2, 2, 3, 3, 4, 4, 5, 5), matcher);
  assertThat(matcherUsed.get(), is(true));
  try {
    CombineFnTester.testCombineFn(
        Sum.ofIntegers(), Arrays.asList(1, 2, 3, 4, 5), Matchers.not(Matchers.equalTo(15)));
  } catch (AssertionError ignored) {
    // Success! Return to avoid the call to fail();
    return;
  }
  fail("The matcher should have failed, throwing an error");
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
/**
 * Tests that if end-of-window and GC timers come in together, that the pane is correctly
 * marked as final.
 */
@Test
public void testCombiningAccumulatingEventTime() throws Exception {
  WindowingStrategy<?, IntervalWindow> strategy =
      WindowingStrategy.of((WindowFn<?, IntervalWindow>) FixedWindows.of(Duration.millis(100)))
          .withTimestampCombiner(TimestampCombiner.EARLIEST)
          .withMode(AccumulationMode.ACCUMULATING_FIRED_PANES)
          .withAllowedLateness(Duration.millis(1))
          .withTrigger(Repeatedly.forever(AfterWatermark.pastEndOfWindow()));

  ReduceFnTester<Integer, Integer, IntervalWindow> tester =
      ReduceFnTester.combining(strategy, Sum.ofIntegers(), VarIntCoder.of());

  injectElement(tester, 2); // processing timer @ 5000 + 10; EOW timer @ 100
  injectElement(tester, 5);

  tester.advanceInputWatermark(new Instant(1000));

  assertThat(
      tester.extractOutput(),
      contains(
          isSingleWindowedValue(
              equalTo(7), 2, 0, 100, PaneInfo.createPane(true, true, Timing.ON_TIME, 0, 0))));
}
 

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

  String instanceId = options.getInstanceId();
  String databaseId = options.getDatabaseId();
  String query = "SELECT * FROM " + options.getTable();

  PCollection<Long> tableEstimatedSize = p
      // Query for all the columns and rows in the specified Spanner table
      .apply(SpannerIO.read()
          .withInstanceId(instanceId)
          .withDatabaseId(databaseId)
          .withQuery(query))
      // Estimate the size of every row
      .apply(ParDo.of(new EstimateStructSizeFn()))
      // Sum all the row sizes to get the total estimated size of the table
      .apply(Sum.longsGlobally());

  // Write the total size to a file
  tableEstimatedSize
      .apply(ToString.elements())
      .apply(TextIO.write().to(options.getOutput()));

  p.run().waitUntilFinish();
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的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.transforms.Sum; //导入依赖的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.transforms.Sum; //导入依赖的package包/类
@Override
public PCollection<KV<String, Integer>> expand(
    PCollection<GameActionInfo> gameInfo) {

  return gameInfo
    .apply(MapElements
        .into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers()))
        .via((GameActionInfo gInfo) -> KV.of(gInfo.getKey(field), gInfo.getScore())))
    .apply(Sum.<String>integersPerKey());
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Override
public PCollection<KV<String, Integer>> expand(PCollection<KV<String, Integer>> userScores) {

  // Get the sum of scores for each user.
  PCollection<KV<String, Integer>> sumScores = userScores
      .apply("UserSum", Sum.<String>integersPerKey());

  // Extract the score from each element, and use it to find the global mean.
  final PCollectionView<Double> globalMeanScore = sumScores.apply(Values.<Integer>create())
      .apply(Mean.<Integer>globally().asSingletonView());

  // Filter the user sums using the global mean.
  PCollection<KV<String, Integer>> filtered = sumScores
      .apply("ProcessAndFilter", ParDo
          // use the derived mean total score as a side input
          .of(new DoFn<KV<String, Integer>, KV<String, Integer>>() {
            private final Counter numSpammerUsers = Metrics.counter("main", "SpammerUsers");
            @ProcessElement
            public void processElement(ProcessContext c) {
              Integer score = c.element().getValue();
              Double gmc = c.sideInput(globalMeanScore);
              if (score > (gmc * SCORE_WEIGHT)) {
                LOG.info("user " + c.element().getKey() + " spammer score " + score
                    + " with mean " + gmc);
                numSpammerUsers.inc();
                c.output(c.element());
              }
            }
          }).withSideInputs(globalMeanScore));
  return filtered;
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
/**
 * {@link CombineFn} for MAX based on {@link Sum} and {@link Combine.BinaryCombineFn}.
 */
public static CombineFn createSum(SqlTypeName fieldType) {
  switch (fieldType) {
    case INTEGER:
      return Sum.ofIntegers();
    case SMALLINT:
      return new ShortSum();
    case TINYINT:
      return new ByteSum();
    case BIGINT:
      return Sum.ofLongs();
    case FLOAT:
      return new FloatSum();
    case DOUBLE:
      return Sum.ofDoubles();
    case DECIMAL:
      return new BigDecimalSum();
    default:
      throw new UnsupportedOperationException(
          String.format("[%s] is not support in SUM", fieldType));
  }
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void testGoodStateParameterSuperclassStateType() throws Exception {
  DoFnSignatures.getSignature(new DoFn<KV<String, Integer>, Long>() {
    @StateId("my-id")
    private final StateSpec<CombiningState<Integer, int[], Integer>> state =
        StateSpecs.combining(Sum.ofIntegers());

    @ProcessElement public void myProcessElement(
        ProcessContext context,
        @StateId("my-id") GroupingState<Integer, Integer> groupingState) {}
  }.getClass());
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void testToFnWithContext() throws Exception {
  CombineFnWithContext<Integer, int[], Integer> fnWithContext =
      CombineFnUtil.toFnWithContext(Sum.ofIntegers());
  List<Integer> inputs = ImmutableList.of(1, 2, 3, 4);
  Context nullContext = CombineContextFactory.nullContext();
  int[] accum = fnWithContext.createAccumulator(nullContext);
  for (Integer i : inputs) {
    accum = fnWithContext.addInput(accum, i, nullContext);
  }
  assertEquals(10, fnWithContext.extractOutput(accum, nullContext).intValue());
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void countAssertsSucceeds() {
  PCollection<Integer> create = pipeline.apply("FirstCreate", Create.of(1, 2, 3));

  PAssert.that(create).containsInAnyOrder(1, 2, 3);
  PAssert.thatSingleton(create.apply(Sum.integersGlobally())).isEqualTo(6);
  PAssert.thatMap(pipeline.apply("CreateMap", Create.of(KV.of(1, 2))))
      .isEqualTo(Collections.singletonMap(1, 2));

  assertThat(PAssert.countAsserts(pipeline), equalTo(3));
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void countAssertsMultipleCallsIndependent() {
  PCollection<Integer> create = pipeline.apply("FirstCreate", Create.of(1, 2, 3));

  PAssert.that(create).containsInAnyOrder(1, 2, 3);
  PAssert.thatSingleton(create.apply(Sum.integersGlobally())).isEqualTo(6);
  assertThat(PAssert.countAsserts(pipeline), equalTo(2));

  PAssert.thatMap(pipeline.apply("CreateMap", Create.of(KV.of(1, 2))))
      .isEqualTo(Collections.singletonMap(1, 2));

  assertThat(PAssert.countAsserts(pipeline), equalTo(3));
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@VisibleForTesting
AdaptiveThrottler(long samplePeriodMs, long sampleUpdateMs,
    double overloadRatio, Random random) {
  allRequests =
      new MovingFunction(samplePeriodMs, sampleUpdateMs,
      1 /* numSignificantBuckets */, 1 /* numSignificantSamples */, Sum.ofLongs());
  successfulRequests =
      new MovingFunction(samplePeriodMs, sampleUpdateMs,
      1 /* numSignificantBuckets */, 1 /* numSignificantSamples */, Sum.ofLongs());
  this.overloadRatio = overloadRatio;
  this.random = random;
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
public MovingAverage(long samplePeriodMs, long sampleUpdateMs,
                      int numSignificantBuckets, int numSignificantSamples) {
  sum = new MovingFunction(samplePeriodMs, sampleUpdateMs,
      numSignificantBuckets, numSignificantSamples, Sum.ofLongs());
  count = new MovingFunction(samplePeriodMs, sampleUpdateMs,
      numSignificantBuckets, numSignificantSamples, Sum.ofLongs());
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Parameters(name = "{index}: {0}")
public static Iterable<Combine.CombineFn<Integer, ?, ?>> params() {
  BinaryCombineIntegerFn sum = Sum.ofIntegers();
  CombineFn<Integer, ?, Long> count = Count.combineFn();
  TestCombineFn test = new TestCombineFn();
  return ImmutableList.<CombineFn<Integer, ?, ?>>builder()
      .add(sum)
      .add(count)
      .add(test)
      .build();
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void writeWithRunnerDeterminedSharding() {
  ResourceId outputDirectory = LocalResources.fromString("/foo/bar", true /* isDirectory */);
  FilenamePolicy policy =
      DefaultFilenamePolicy.fromStandardParameters(
          StaticValueProvider.of(outputDirectory),
          DefaultFilenamePolicy.DEFAULT_UNWINDOWED_SHARD_TEMPLATE,
          "",
          false);
  WriteFiles<Integer, Void, Integer> write =
      WriteFiles.to(
          new FileBasedSink<Integer, Void, Integer>(
              StaticValueProvider.of(outputDirectory),
              DynamicFileDestinations.<Integer>constant(new FakeFilenamePolicy())) {
            @Override
            public WriteOperation<Void, Integer> createWriteOperation() {
              return null;
            }
          });
  assertThat(
      PTransformMatchers.writeWithRunnerDeterminedSharding().matches(appliedWrite(write)),
      is(true));

  WriteFiles<Integer, Void, Integer> withStaticSharding = write.withNumShards(3);
  assertThat(
      PTransformMatchers.writeWithRunnerDeterminedSharding()
          .matches(appliedWrite(withStaticSharding)),
      is(false));

  WriteFiles<Integer, Void, Integer> withCustomSharding =
      write.withSharding(Sum.integersGlobally().asSingletonView());
  assertThat(
      PTransformMatchers.writeWithRunnerDeterminedSharding()
          .matches(appliedWrite(withCustomSharding)),
      is(false));
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Before
public void setup() {
  MockitoAnnotations.initMocks(this);
  PCollection<Integer> created = p.apply(Create.of(1, 2, 3));
  singletonView =
      created
          .apply(Window.into(new IdentitySideInputWindowFn()))
          .apply(Sum.integersGlobally().asSingletonView());

  underlying = new TestDoFnRunner<>();
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
/**
 * Tests that a processing time timer does not cause window GC.
 */
@Test
public void testProcessingTimeTimerDoesNotGc() throws Exception {
  WindowingStrategy<?, IntervalWindow> strategy =
      WindowingStrategy.of((WindowFn<?, IntervalWindow>) FixedWindows.of(Duration.millis(100)))
          .withTimestampCombiner(TimestampCombiner.EARLIEST)
          .withMode(AccumulationMode.ACCUMULATING_FIRED_PANES)
          .withAllowedLateness(Duration.ZERO)
          .withTrigger(
              Repeatedly.forever(
                  AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(10))));

  ReduceFnTester<Integer, Integer, IntervalWindow> tester =
      ReduceFnTester.combining(strategy, Sum.ofIntegers(), VarIntCoder.of());

  tester.advanceProcessingTime(new Instant(5000));
  injectElement(tester, 2); // processing timer @ 5000 + 10; EOW timer @ 100
  injectElement(tester, 5);

  tester.advanceProcessingTime(new Instant(10000));

  tester.assertHasOnlyGlobalAndStateFor(
      new IntervalWindow(new Instant(0), new Instant(100)));

  assertThat(
      tester.extractOutput(),
      contains(
          isSingleWindowedValue(
              equalTo(7), 2, 0, 100, PaneInfo.createPane(true, false, Timing.EARLY, 0, 0))));
}
 

import org.apache.beam.sdk.transforms.Sum; //导入依赖的package包/类
@Test
public void testOnElementCombiningDiscarding() throws Exception {
  // Test basic execution of a trigger using a non-combining window set and discarding mode.

  WindowingStrategy<?, IntervalWindow> strategy =
      WindowingStrategy.of((WindowFn<?, IntervalWindow>) FixedWindows.of(Duration.millis(10)))
          .withTimestampCombiner(TimestampCombiner.EARLIEST)
          .withMode(AccumulationMode.DISCARDING_FIRED_PANES)
          .withAllowedLateness(Duration.millis(100));

  ReduceFnTester<Integer, Integer, IntervalWindow> tester =
      ReduceFnTester.combining(
          strategy,
          mockTriggerStateMachine,
          Sum.ofIntegers(),
          VarIntCoder.of());

  injectElement(tester, 2);

  when(mockTriggerStateMachine.shouldFire(anyTriggerContext())).thenReturn(true);
  injectElement(tester, 3);

  when(mockTriggerStateMachine.shouldFire(anyTriggerContext())).thenReturn(true);
  triggerShouldFinish(mockTriggerStateMachine);
  injectElement(tester, 4);

  // This element shouldn't be seen, because the trigger has finished
  injectElement(tester, 6);

  assertThat(
      tester.extractOutput(),
      contains(
          isSingleWindowedValue(equalTo(5), 2, 0, 10),
          isSingleWindowedValue(equalTo(4), 4, 0, 10)));
  assertTrue(tester.isMarkedFinished(firstWindow));
  tester.assertHasOnlyGlobalAndFinishedSetsFor(firstWindow);
}