View { public static <K, V> AsMultimap<K, V> asMultimap() { return new AsMultimap<>(); } private  View(); static AsSingleton<T> asSingleton(); static AsList<T> asList(); static AsIterable<T> asIterable(); static AsMap<K, V> asMap(); static AsMultimap<K, V> asMultimap();  }
@Test @Category(ValidatesRunner.class) public void testMultimapSideInput() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testMultimapAsEntrySetSideInput() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply("CreateMainInput", Create.of(2 )) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { assertEquals((int) c.element(), c.sideInput(view).size()); assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); for (Entry<String, Iterable<Integer>> entry : c.sideInput(view).entrySet()) { for (Integer value : entry.getValue()) { c.output(KV.of(entry.getKey(), value)); } } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); pipeline.run(); }
@Test @Category({ValidatesRunner.class}) public void testMultimapSideInputWithNonDeterministicKeyCoder() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)) .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedMultimapSideInput() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(KV.of("a", 1), new Instant(1)), TimestampedValue.of(KV.of("a", 1), new Instant(2)), TimestampedValue.of(KV.of("a", 2), new Instant(7)), TimestampedValue.of(KV.of("b", 3), new Instant(14)))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of("apple", new Instant(5)), TimestampedValue.of("banana", new Instant(13)), TimestampedValue.of("blackberry", new Instant(16)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testWindowedMultimapAsEntrySetSideInput() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(KV.of("a", 1), new Instant(1)), TimestampedValue.of(KV.of("a", 1), new Instant(2)), TimestampedValue.of(KV.of("a", 2), new Instant(7)), TimestampedValue.of(KV.of("b", 3), new Instant(14)))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of(1 , new Instant(5)), TimestampedValue.of(1 , new Instant(16)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { assertEquals((int) c.element(), c.sideInput(view).size()); assertEquals((int) c.element(), c.sideInput(view).entrySet().size()); for (Entry<String, Iterable<Integer>> entry : c.sideInput(view).entrySet()) { for (Integer value : entry.getValue()) { c.output(KV.of(entry.getKey(), value)); } } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)); pipeline.run(); }
@Test @Category({ValidatesRunner.class}) public void testWindowedMultimapSideInputWithNonDeterministicKeyCoder() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(KV.of("a", 1), new Instant(1)), TimestampedValue.of(KV.of("a", 1), new Instant(2)), TimestampedValue.of(KV.of("a", 2), new Instant(7)), TimestampedValue.of(KV.of("b", 3), new Instant(14))) .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of("apple", new Instant(5)), TimestampedValue.of("banana", new Instant(13)), TimestampedValue.of("blackberry", new Instant(16)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testEmptyMultimapSideInput() throws Exception { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateEmptyView", Create.empty(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()))) .apply(View.asMultimap()); PCollection<Integer> results = pipeline .apply("Create1", Create.of(1)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { assertTrue(c.sideInput(view).isEmpty()); assertTrue(c.sideInput(view).entrySet().isEmpty()); assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); c.output(c.element()); } }) .withSideInputs(view)); PAssert.that(results).containsInAnyOrder(1); pipeline.run(); }
@Test @Category({ValidatesRunner.class}) public void testEmptyMultimapSideInputWithNonDeterministicKeyCoder() throws Exception { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateEmptyView", Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply(View.asMultimap()); PCollection<Integer> results = pipeline .apply("Create1", Create.of(1)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { assertTrue(c.sideInput(view).isEmpty()); assertTrue(c.sideInput(view).entrySet().isEmpty()); assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); c.output(c.element()); } }) .withSideInputs(view)); PAssert.that(results).containsInAnyOrder(1); pipeline.run(); }
@Test @Category(ValidatesRunner.class) public void testMultimapSideInputIsImmutable() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline.apply("CreateSideInput", Create.of(KV.of("a", 1))).apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply("CreateMainInput", Create.of("apple")) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { try { c.sideInput(view).clear(); fail("Expected UnsupportedOperationException on clear()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).put("c", ImmutableList.of(3)); fail("Expected UnsupportedOperationException on put()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).remove("c"); fail("Expected UnsupportedOperationException on remove()"); } catch (UnsupportedOperationException expected) { } try { c.sideInput(view).putAll(new HashMap<>()); fail("Expected UnsupportedOperationException on putAll()"); } catch (UnsupportedOperationException expected) { } for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output).containsInAnyOrder(KV.of("apple", 1)); pipeline.run(); }
@Test public void testViewUnboundedAsMultimapDirect() { testViewUnbounded(pipeline, View.asMultimap()); }
@Test public void testViewNonmergingAsMultimapDirect() { testViewNonmerging(pipeline, View.asMultimap()); }
@Test @Category({ValidatesRunner.class, DataflowPortabilityApiUnsupported.class}) public void testMultimapSideInputWithNonDeterministicKeyCoder() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.of(KV.of("a", 1), KV.of("a", 1), KV.of("a", 2), KV.of("b", 3)) .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply("CreateMainInput", Create.of("apple", "banana", "blackberry")) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category({ValidatesRunner.class, DataflowPortabilityApiUnsupported.class}) public void testWindowedMultimapSideInputWithNonDeterministicKeyCoder() { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateSideInput", Create.timestamped( TimestampedValue.of(KV.of("a", 1), new Instant(1)), TimestampedValue.of(KV.of("a", 1), new Instant(2)), TimestampedValue.of(KV.of("a", 2), new Instant(7)), TimestampedValue.of(KV.of("b", 3), new Instant(14))) .withCoder(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply("SideWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply(View.asMultimap()); PCollection<KV<String, Integer>> output = pipeline .apply( "CreateMainInput", Create.timestamped( TimestampedValue.of("apple", new Instant(5)), TimestampedValue.of("banana", new Instant(13)), TimestampedValue.of("blackberry", new Instant(16)))) .apply("MainWindowInto", Window.into(FixedWindows.of(Duration.millis(10)))) .apply( "OutputSideInputs", ParDo.of( new DoFn<String, KV<String, Integer>>() { @ProcessElement public void processElement(ProcessContext c) { for (Integer v : c.sideInput(view).get(c.element().substring(0, 1))) { c.output(KV.of(c.element(), v)); } } }) .withSideInputs(view)); PAssert.that(output) .containsInAnyOrder( KV.of("apple", 1), KV.of("apple", 1), KV.of("apple", 2), KV.of("banana", 3), KV.of("blackberry", 3)); pipeline.run(); }
@Test @Category({ValidatesRunner.class, DataflowPortabilityApiUnsupported.class}) public void testEmptyMultimapSideInputWithNonDeterministicKeyCoder() throws Exception { final PCollectionView<Map<String, Iterable<Integer>>> view = pipeline .apply( "CreateEmptyView", Create.empty(KvCoder.of(new NonDeterministicStringCoder(), VarIntCoder.of()))) .apply(View.asMultimap()); PCollection<Integer> results = pipeline .apply("Create1", Create.of(1)) .apply( "OutputSideInputs", ParDo.of( new DoFn<Integer, Integer>() { @ProcessElement public void processElement(ProcessContext c) { assertTrue(c.sideInput(view).isEmpty()); assertTrue(c.sideInput(view).entrySet().isEmpty()); assertFalse(c.sideInput(view).entrySet().iterator().hasNext()); c.output(c.element()); } }) .withSideInputs(view)); PAssert.that(results).containsInAnyOrder(1); pipeline.run(); }