Create { public static <T> Values<T> of(Iterable<T> elems) { return new Values<>(elems, Optional.absent(), Optional.absent()); }  static Values<T> of(Iterable<T> elems); @SafeVarargs static Values<T> of(@Nullable T elem, @Nullable T... elems); @Experimental(Kind.SCHEMAS) static Values<Row> empty(Schema schema); static Values<T> empty(Coder<T> coder); static Values<T> empty(TypeDescriptor<T> type); static Values<KV<K, V>> of(Map<K, V> elems); static OfValueProvider<T> ofProvider(ValueProvider<T> provider, Coder<T> coder); static TimestampedValues<T> timestamped(Iterable<TimestampedValue<T>> elems); @SafeVarargs static TimestampedValues<T> timestamped(
      TimestampedValue<T> elem, @SuppressWarnings("unchecked") TimestampedValue<T>... elems); static TimestampedValues<T> timestamped(
      Iterable<T> values, Iterable<Long> timestamps);  }
@Test @Category(ValidatesRunner.class) public void testCreate() { PCollection<String> output = p.apply(Create.of(LINES)); PAssert.that(output).containsInAnyOrder(LINES_ARRAY); p.run(); }
@Test public void testCreateEmptyIterableRequiresCoder() { p.enableAbandonedNodeEnforcement(false); thrown.expect(IllegalArgumentException.class); thrown.expectMessage("determine a default Coder"); thrown.expectMessage("Create.empty(Coder)"); thrown.expectMessage("Create.empty(TypeDescriptor)"); thrown.expectMessage("withCoder(Coder)"); thrown.expectMessage("withType(TypeDescriptor)"); p.apply(Create.of(Collections.emptyList())); }
@Test public void testPolymorphicType() throws Exception { thrown.expect(RuntimeException.class); thrown.expectMessage(Matchers.containsString("Unable to infer a coder")); p.apply(Create.of(new Record(), new Record2())); p.run(); }
@Test @Category(ValidatesRunner.class) public void testCreateWithNullsAndValues() throws Exception { PCollection<String> output = p.apply( Create.of(null, "test1", null, "test2", null) .withCoder(SerializableCoder.of(String.class))); PAssert.that(output).containsInAnyOrder(null, "test1", null, "test2", null); p.run(); }
@Test @Category(NeedsRunner.class) public void testCreateParameterizedType() throws Exception { PCollection<TimestampedValue<String>> output = p.apply( Create.of( TimestampedValue.of("a", new Instant(0)), TimestampedValue.of("b", new Instant(0)))); PAssert.that(output) .containsInAnyOrder( TimestampedValue.of("a", new Instant(0)), TimestampedValue.of("b", new Instant(0))); p.run(); }
@Test @Category(NeedsRunner.class) public void testCreateWithUnserializableElements() throws Exception { List<UnserializableRecord> elements = ImmutableList.of( new UnserializableRecord("foo"), new UnserializableRecord("bar"), new UnserializableRecord("baz")); Create.Values<UnserializableRecord> create = Create.of(elements).withCoder(new UnserializableRecord.UnserializableRecordCoder()); PAssert.that(p.apply(create)) .containsInAnyOrder( new UnserializableRecord("foo"), new UnserializableRecord("bar"), new UnserializableRecord("baz")); p.run(); }
@Test @Category(ValidatesRunner.class) public void testCreateWithVoidType() throws Exception { PCollection<Void> output = p.apply(Create.of((Void) null, (Void) null)); PAssert.that(output).containsInAnyOrder((Void) null, (Void) null); p.run(); }
@Test @Category(ValidatesRunner.class) public void testCreateWithKVVoidType() throws Exception { PCollection<KV<Void, Void>> output = p.apply(Create.of(KV.of((Void) null, (Void) null), KV.of((Void) null, (Void) null))); PAssert.that(output) .containsInAnyOrder(KV.of((Void) null, (Void) null), KV.of((Void) null, (Void) null)); p.run(); }
@Test public void testCreateDefaultOutputCoderUsingInference() throws Exception { Coder<Record> coder = new RecordCoder(); p.getCoderRegistry().registerCoderForClass(Record.class, coder); assertThat( p.apply(Create.of(new Record(), new Record(), new Record())).getCoder(), equalTo(coder)); }
@Test public void testCreateDefaultOutputCoderUsingCoder() throws Exception { Coder<Record> coder = new RecordCoder(); assertThat( p.apply(Create.of(new Record(), new Record2()).withCoder(coder)).getCoder(), equalTo(coder)); }
@Test public void testCreateDefaultOutputCoderUsingTypeDescriptor() throws Exception { Coder<Record> coder = new RecordCoder(); p.getCoderRegistry().registerCoderForClass(Record.class, coder); Create.Values<Record> values = Create.of(new Record(), new Record2()).withType(new TypeDescriptor<Record>() {}); assertThat(p.apply(values).getCoder(), equalTo(coder)); }
@Test public void testCreateRegisteredSchema() { p.getSchemaRegistry() .registerSchemaForClass( String.class, STRING_SCHEMA, s -> Row.withSchema(STRING_SCHEMA).addValue(s).build(), r -> r.getString("field")); PCollection<String> out = p.apply(Create.of("a", "b", "c", "d")); assertThat(out.getCoder(), instanceOf(SchemaCoder.class)); }
@Test public void testCreateExplicitSchema() { PCollection<String> out = p.apply( Create.of("a", "b", "c", "d") .withSchema( STRING_SCHEMA, TypeDescriptors.strings(), s -> Row.withSchema(STRING_SCHEMA).addValue(s).build(), r -> r.getString("field"))); assertThat(out.getCoder(), instanceOf(SchemaCoder.class)); }
@Test public void testSourceIsSerializableWithUnserializableElements() throws Exception { List<UnserializableRecord> elements = ImmutableList.of( new UnserializableRecord("foo"), new UnserializableRecord("bar"), new UnserializableRecord("baz")); CreateSource<UnserializableRecord> source = CreateSource.fromIterable(elements, new UnserializableRecord.UnserializableRecordCoder()); SerializableUtils.ensureSerializable(source); }
@Test public void testSourceSplit() throws Exception { CreateSource<Integer> source = CreateSource.fromIterable( ImmutableList.of(1, 2, 3, 4, 5, 6, 7, 8), BigEndianIntegerCoder.of()); PipelineOptions options = PipelineOptionsFactory.create(); List<? extends BoundedSource<Integer>> splitSources = source.split(12, options); assertThat(splitSources, hasSize(3)); SourceTestUtils.assertSourcesEqualReferenceSource(source, splitSources, options); }
@Test public void testSourceSplitVoid() throws Exception { CreateSource<Void> source = CreateSource.fromIterable(Lists.newArrayList(null, null, null, null, null), VoidCoder.of()); PipelineOptions options = PipelineOptionsFactory.create(); List<? extends BoundedSource<Void>> splitSources = source.split(3, options); SourceTestUtils.assertSourcesEqualReferenceSource(source, splitSources, options); }
@Test public void testSourceSplitEmpty() throws Exception { CreateSource<Integer> source = CreateSource.fromIterable(ImmutableList.of(), BigEndianIntegerCoder.of()); PipelineOptions options = PipelineOptionsFactory.create(); List<? extends BoundedSource<Integer>> splitSources = source.split(12, options); SourceTestUtils.assertSourcesEqualReferenceSource(source, splitSources, options); }
@Test public void testSourceGetOutputCoderReturnsConstructorCoder() throws Exception { Coder<Integer> coder = VarIntCoder.of(); CreateSource<Integer> source = CreateSource.fromIterable(ImmutableList.of(1, 2, 3, 4, 5, 6, 7, 8), coder); Coder<Integer> defaultCoder = source.getOutputCoder(); assertThat(defaultCoder, equalTo(coder)); }
@Test public void testSourceSplitAtFraction() throws Exception { List<Integer> elements = new ArrayList<>(); Random random = new Random(); for (int i = 0; i < 25; i++) { elements.add(random.nextInt()); } CreateSource<Integer> source = CreateSource.fromIterable(elements, VarIntCoder.of()); SourceTestUtils.assertSplitAtFractionExhaustive(source, PipelineOptionsFactory.create()); }