Hello all - I am trying to run a barebone beam pipeline to understand the "combine" logic. I am from python world trying to learn java beam sdk due to my use case of ETL with spark cluster. So, pardon me for my grotesque java code :)
I appreciate if you could nudge me in the right path with this error: (please see below) Here's my code: (read lines from input file and output the same lines to outfile) public class VariantCaller { public static void main( String[] args ) { PipelineOptions opts = PipelineOptionsFactory.fromArgs(args ).create(); Pipeline p = Pipeline.create(opts); PCollection<String> lines = p.apply(TextIO.read().from("test_in.csv" )); PCollection<String> mergedLines = lines.apply(Combine.globally(new AddLines())) mergedLines.apply(TextIO.write().to("test_out.csv")); p.run(); } } AddLines Class: public class AddLines extends CombineFn<String, AddLines.Accum, String> { /** * */ private static final long serialVersionUID = 1L; public static class Accum { String line; } @Override public Accum createAccumulator() { return new Accum(); } @Override public Accum addInput(Accum accum, String line) { accum.line.concat(line); return accum; } @Override public Accum mergeAccumulators(Iterable<Accum> accums) { Accum merged = createAccumulator(); for (Accum accum : accums) { merged.line.concat("\n").concat(accum.line); } return merged; } @Override public String extractOutput(Accum accum) { return accum.line; } } Exception in thread "main" java.lang.IllegalStateException: Unable to return a default Coder for Combine.globally(AddLines)/Combine.perKey(AddLines)/Combine.GroupedValues/ParDo(Anonymous)/ParMultiDo(Anonymous).output [PCollection]. Correct one of the following root causes: No Coder has been manually specified; you may do so using .setCoder(). Inferring a Coder from the CoderRegistry failed: Cannot provide coder for parameterized type org.apache.beam.sdk.values.KV<K, OutputT>: Unable to provide a Coder for K. Building a Coder using a registered CoderProvider failed. See suppressed exceptions for detailed failures. Using the default output Coder from the producing PTransform failed: PTransform.getOutputCoder called. at org.apache.beam.repackaged.beam_sdks_java_core.com.google.common.base.Preconditions.checkState( Preconditions.java:444) at org.apache.beam.sdk.values.PCollection.getCoder(PCollection.java:277) at org.apache.beam.sdk.values.PCollection.finishSpecifying( PCollection.java:114) at org.apache.beam.sdk.runners.TransformHierarchy.finishSpecifyingInput( TransformHierarchy.java:190) at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:536) at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:471) at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:325) at org.apache.beam.sdk.transforms.Combine$Globally.expand(Combine.java:1074) at org.apache.beam.sdk.transforms.Combine$Globally.expand(Combine.java:943) at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:537) at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:471) at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:325) at pipelines.variant_caller.VariantCaller.main(VariantCaller.java:27) *--* *Mahesh Vangala* *(Ph) 443-326-1957* *(web) mvangala.com <http://mvangala.com>*