Dataflow job template creation errors out

Question

I want to create a dataflow job template that takes the filename in GCS and publishes it to a PubSub topic. I followed the tutorial at this link, but that doesn't seem to be working for me.

My class definition is the following -

import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.io.PubsubIO;
import com.google.cloud.dataflow.sdk.io.TextIO;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.runners.TemplatingDataflowPipelineRunner;

public class PubSubOutputTest {
  public static void main(String[] args) {
    // Create pipeline options.
    pubSubOutputOptions options = PipelineOptionsFactory.fromArgs(args).as(pubSubOutputOptions.class);
    options.setRunner(TemplatingDataflowPipelineRunner.class);
    options.setTempLocation("gs://staging-bucket");
    Pipeline p = Pipeline.create(options);
    // Read the file from the GCS Bucket.
    p.apply(TextIO.Read.named("Read file from GCS.").from(options.getInputFile()).withoutValidation())
        .apply(PubsubIO.Write.named("Write to Pub Sub topic.")
        .topic("projects/my-project/topics/my-topic"));
    // Run the pipeline.
    p.run();
  }
}

The interface that implements the ValueProvider to grab the runtime inputs is the following -

import com.google.cloud.dataflow.sdk.options.Default;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.ValueProvider;

public interface pubSubOutputOptions extends PipelineOptions {
  @Default.String("gs://default-file.txt")
  ValueProvider getInputFile();
  void setInputFile(ValueProvider value);
}

The template creation is giving the following error.

Exception in thread "main" java.lang.IllegalArgumentException: PipelineOptions specified failed to serialize to JSON.
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineTranslator$Translator.translate(DataflowPipelineTranslator.java:408)
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineTranslator.translate(DataflowPipelineTranslator.java:146)
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.run(DataflowPipelineRunner.java:570)
    at com.google.cloud.dataflow.sdk.runners.TemplatingDataflowPipelineRunner.run(TemplatingDataflowPipelineRunner.java:137)
    at com.google.cloud.dataflow.sdk.runners.TemplatingDataflowPipelineRunner.run(TemplatingDataflowPipelineRunner.java:44)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at com.my.project.dataflow.PubSubOutputTest.main(PubSubOutputTest.java:32)
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Unexpected IOException (of type java.io.IOException): Failed to serialize and deserialize property 'inputFile' with value 'RuntimeValueProvider{propertyName=inputFile, default=gs://default-file.txt, value=null}'
    at com.fasterxml.jackson.databind.JsonMappingException.fromUnexpectedIOE(JsonMappingException.java:284)
    at com.fasterxml.jackson.databind.ObjectMapper.writeValueAsBytes(ObjectMapper.java:3008)
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineTranslator$Translator.translate(DataflowPipelineTranslator.java:406)

I am new to Google Cloud Dataflow and Java. I implemented everything in the documentation but I could have missed something obvious.

Sam McVeety Sam McVeety · Accepted Answer · 2017-02-01T17:42:30

It looks like there's an error with how you're declaring the option. I think you want to provide a template parameter to ValueProvider, like so:

  @Default.String("gs://default-file.txt")
  ValueProvider<String> getInputFile();
  void setInputFile(ValueProvider<String> value);

Dataflow job template creation errors out

1 Answers