Running java dataflow pipeline locally with PubSub

Question

I'm trying to run locally (with local runner), in Eclipse, a Java dataflow which use Apache Beam.

The problem occurs when I added pubsub as the data source.
(If I delete that line and try with a simple pipeline, I can execute it without problem)

The pipeline code:

public class StarterPipeline {

 public static void main(String[] args) {
    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
    run(options);
  }

public static PipelineResult run(Options options) {

    Pipeline pipeline = Pipeline.create(options);

    /*
     * Step 1: Read from PubSub
     */
    PCollection<PubsubMessage> messages = null;
    if (options.getUseSubscription()) {
        messages = pipeline.apply("ReadPubSubSubscription", PubsubIO.readMessagesWithAttributes()
                .fromSubscription(options.getInputSubscription()).withIdAttribute("messageId"));
    } else {
        messages = pipeline.apply("ReadPubSubTopic", PubsubIO.readMessagesWithAttributes()
                .fromTopic(options.getInputTopic()).withIdAttribute("messageId"));
    }

     <...>
    return pipeline.run();
}

And the error message:

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/beam/sdk/util/DoFnAndMainOutput
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.translateDoFn(ParDoTranslation.java:462)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$1.translateDoFn(ParDoTranslation.java:160)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.payloadForParDoLike(ParDoTranslation.java:695)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:156)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$ParDoPayloadTranslator.translate(ParDoTranslation.java:111)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:206)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:547)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:557)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformMatchers$4.matches(PTransformMatchers.java:194)
    at org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:282)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:665)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:317)
    at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:251)
    at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:460)
    at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:260)
    at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:210)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:170)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:66)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:315)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:301)
    at com.accenture.pipeline.StarterPipeline.run(StarterPipeline.java:128)
    at com.accenture.pipeline.StarterPipeline.main(StarterPipeline.java:74)
Caused by: java.lang.ClassNotFoundException: org.apache.beam.sdk.util.DoFnAndMainOutput
    at java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:602)
    at java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:521)
    ... 26 more

To execute the pipeline locally I have tried to run with:

Maven
Dataflow SDK

With the following options:

DATAFLOW

MAVEN

Should I add any dependency?

If you need any further details, let me know.

Looking at the stack trace, seems like you are using DirectRunnner not DataflowRunner. Can you refer to following to properly setup your Dataflow pipeline ? cloud.google.com/dataflow/docs/quickstarts/… Unfortunately seems like provided eclipse plugin does not work for latest versions of Beam if you were trying to use that. — chamikara

IoT user IoT user · Accepted Answer · 2019-12-05T07:12:28

To solve that, I added the following dependency:

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-direct-java</artifactId>
  <version>${beam.version}</version>
  <scope>runtime</scope>
</dependency>

Running java dataflow pipeline locally with PubSub

1 Answers