MongoDbGridFSIO { public static Read<String> read() { return new AutoValue_MongoDbGridFSIO_Read.Builder<String>() .setParser(TEXT_PARSER) .setCoder(StringUtf8Coder.of()) .setConnectionConfiguration(ConnectionConfiguration.create()) .setSkew(Duration.ZERO) .build(); }  static Read<String> read(); static Write<String> write(); static Write<T> write(WriteFn<T> fn);  }
@Test public void testFullRead() { PCollection<String> output = pipeline.apply( MongoDbGridFSIO.read().withUri("mongodb: PAssert.thatSingleton(output.apply("Count All", Count.globally())).isEqualTo(5000L); PAssert.that(output.apply("Count PerElement", Count.perElement())) .satisfies( input -> { for (KV<String, Long> element : input) { assertEquals(500L, element.getValue().longValue()); } return null; }); pipeline.run(); }
@Test public void testReadWithParser() { PCollection<KV<String, Integer>> output = pipeline.apply( MongoDbGridFSIO.read() .withUri("mongodb: .withDatabase(DATABASE) .withBucket("mapBucket") .<KV<String, Integer>>withParser( (input, callback) -> { try (final BufferedReader reader = new BufferedReader( new InputStreamReader( input.getInputStream(), StandardCharsets.UTF_8))) { String line = reader.readLine(); while (line != null) { try (Scanner scanner = new Scanner(line.trim())) { scanner.useDelimiter("\\t"); long timestamp = scanner.nextLong(); String name = scanner.next(); int score = scanner.nextInt(); callback.output(KV.of(name, score), new Instant(timestamp)); } line = reader.readLine(); } } }) .withSkew(new Duration(3610000L)) .withCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))); PAssert.thatSingleton(output.apply("Count All", Count.globally())).isEqualTo(50100L); PAssert.that(output.apply("Max PerElement", Max.integersPerKey())) .satisfies( input -> { for (KV<String, Integer> element : input) { assertEquals(101, element.getValue().longValue()); } return null; }); pipeline.run(); }
@Test public void testSplit() throws Exception { PipelineOptions options = PipelineOptionsFactory.create(); MongoDbGridFSIO.Read<String> read = MongoDbGridFSIO.read().withUri("mongodb: BoundedGridFSSource src = new BoundedGridFSSource(read, null); long desiredBundleSizeBytes = (src.getEstimatedSizeBytes(options) * 2L) / 5L + 1000; List<? extends BoundedSource<ObjectId>> splits = src.split(desiredBundleSizeBytes, options); int expectedNbSplits = 3; assertEquals(expectedNbSplits, splits.size()); SourceTestUtils.assertSourcesEqualReferenceSource(src, splits, options); int nonEmptySplits = 0; int count = 0; for (BoundedSource<ObjectId> subSource : splits) { List<ObjectId> result = SourceTestUtils.readFromSource(subSource, options); if (result.size() > 0) { nonEmptySplits += 1; } count += result.size(); } assertEquals(expectedNbSplits, nonEmptySplits); assertEquals(5, count); }