@Test public void processorBulk() { final List<Sample> data = IntStream .range(0, 1000) .mapToObj(i -> IntStream.range(0, i).mapToObj(j -> "a").collect(joining(""))) .map(Sample::new) .collect(toList()); final PCollection<Integer> out = pipeline .apply(UUID.randomUUID().toString(), Create.of(data).withCoder(JsonbCoder.of(Sample.class, PLUGIN))) .apply(UUID.randomUUID().toString(), toRecord()) .setCoder(SchemaRegistryCoder.of()) .apply(UUID.randomUUID().toString(), RecordNormalizer.of(PLUGIN)) .apply(UUID.randomUUID().toString(), TalendFn.asFn(new BaseTestProcessor() { private final Collection<Sample> objects = new ArrayList<>(); @Override public void beforeGroup() { objects.clear(); } @Override public void afterGroup(final OutputFactory output) { objects.forEach(output.create(Branches.DEFAULT_BRANCH)::emit); } @Override public void onNext(final InputFactory input, final OutputFactory factory) { objects .add(new Sample( Record.class.cast(input.read(Branches.DEFAULT_BRANCH)).getString("data"))); } })) .setCoder(SchemaRegistryCoder.of()) .apply(UUID.randomUUID().toString(), toSampleLength()) .apply(UUID.randomUUID().toString(), toInt()); final List<Integer> expected = data.stream().map(Sample::getData).map(String::length).collect(toList()); PAssert.that(out).satisfies((SerializableFunction<Iterable<Integer>, Void>) input -> { final List<Integer> actual = StreamSupport.stream(input.spliterator(), false).sorted().collect(toList()); assertEquals(expected, actual); return null; }); assertEquals(PipelineResult.State.DONE, pipeline.run().getState()); }
public static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper) { return read(mapper, emptyMap()); }
TalendIO { public static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper) { return read(mapper, emptyMap()); } }
TalendIO { public static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper) { return read(mapper, emptyMap()); }  }
TalendIO { public static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper) { return read(mapper, emptyMap()); }  static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper); static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper,
            final Map<String, String> mapperConfiguration); static Write write(final Processor output); }
TalendIO { public static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper) { return read(mapper, emptyMap()); }  static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper); static Base<PBegin, PCollection<Record>, Mapper> read(final Mapper mapper,
            final Map<String, String> mapperConfiguration); static Write write(final Processor output);  }