CountingSource { @Deprecated public static BoundedSource<Long> upTo(long numElements) { checkArgument( numElements >= 0, "numElements (%s) must be greater than or equal to 0", numElements); return new BoundedCountingSource(0, numElements); } private  CountingSource(); @Deprecated static BoundedSource<Long> upTo(long numElements); @Deprecated static UnboundedSource<Long, CounterMark> unbounded(); @Deprecated static UnboundedSource<Long, CounterMark> unboundedWithTimestampFn(
      SerializableFunction<Long, Instant> timestampFn);  }
@Test @Category(NeedsRunner.class) public void testBoundedSource() { long numElements = 1000; PCollection<Long> input = p.apply(Read.from(CountingSource.upTo(numElements))); addCountingAsserts(input, numElements); p.run(); }
@Test @Category(NeedsRunner.class) public void testEmptyBoundedSource() { PCollection<Long> input = p.apply(Read.from(CountingSource.upTo(0))); PAssert.that(input).empty(); p.run(); }
@Test @Category({ ValidatesRunner.class, UsesStatefulParDo.class }) public void testBoundedSourceSplits() throws Exception { long numElements = 1000; long numSplits = 10; long splitSizeBytes = numElements * 8 / numSplits; BoundedSource<Long> initial = CountingSource.upTo(numElements); List<? extends BoundedSource<Long>> splits = initial.split(splitSizeBytes, p.getOptions()); assertEquals("Expected exact splitting", numSplits, splits.size()); PCollectionList<Long> pcollections = PCollectionList.empty(p); for (int i = 0; i < splits.size(); ++i) { BoundedSource<Long> split = splits.get(i); pcollections = pcollections.and(p.apply("split" + i, Read.from(split))); assertEquals( "Expected even splitting", splitSizeBytes, split.getEstimatedSizeBytes(p.getOptions())); } PCollection<Long> input = pcollections.apply(Flatten.pCollections()); addCountingAsserts(input, numElements); p.run(); }
@Test public void testProgress() throws IOException { final int numRecords = 5; @SuppressWarnings("deprecation") BoundedSource<Long> source = CountingSource.upTo(numRecords); try (BoundedReader<Long> reader = source.createReader(PipelineOptionsFactory.create())) { assertEquals(0.0, reader.getFractionConsumed(), 1e-6); assertEquals(0, reader.getSplitPointsConsumed()); assertEquals(numRecords, reader.getSplitPointsRemaining()); assertTrue(reader.start()); int i = 0; do { assertEquals(i, reader.getSplitPointsConsumed()); assertEquals(numRecords - i, reader.getSplitPointsRemaining()); ++i; } while (reader.advance()); assertEquals(numRecords, i); assertEquals(1.0, reader.getFractionConsumed(), 1e-6); assertEquals(numRecords, reader.getSplitPointsConsumed()); assertEquals(0, reader.getSplitPointsRemaining()); } }
@Test @Category({ ValidatesRunner.class, UsesStatefulParDo.class, DataflowPortabilityApiUnsupported.class }) public void testBoundedSourceSplits() throws Exception { long numElements = 1000; long numSplits = 10; long splitSizeBytes = numElements * 8 / numSplits; BoundedSource<Long> initial = CountingSource.upTo(numElements); List<? extends BoundedSource<Long>> splits = initial.split(splitSizeBytes, p.getOptions()); assertEquals("Expected exact splitting", numSplits, splits.size()); PCollectionList<Long> pcollections = PCollectionList.empty(p); for (int i = 0; i < splits.size(); ++i) { BoundedSource<Long> split = splits.get(i); pcollections = pcollections.and(p.apply("split" + i, Read.from(split))); assertEquals( "Expected even splitting", splitSizeBytes, split.getEstimatedSizeBytes(p.getOptions())); } PCollection<Long> input = pcollections.apply(Flatten.pCollections()); addCountingAsserts(input, numElements); p.run(); }