Cloud Dataflow Custom Template creation Issues

Question

I am trying to create a template for cloud data flow job that reads json file from cloud storage and writes to Big Query. I am passing 2 runtime arguments : 1. InputFile for GCS location 2. Dataset and Table Id of BigQuery.

JsonTextToBqTemplate code:

 public class JsonTextToBqTemplate {

    private static final Logger logger = 
    LoggerFactory.getLogger(TextToBQTemplate.class);

    private static Gson gson = new GsonBuilder().create();

    public static void main(String[] args) throws Exception {

        JsonToBQTemplateOptions options = 
        PipelineOptionsFactory.fromArgs(args).withValidation()
                .as(JsonToBQTemplateOptions.class);

        String jobName = options.getJobName();

        try {
            logger.info("PIPELINE-INFO: jobName={} message={} ", 
            jobName, "starting pipeline creation");
            Pipeline pipeline = Pipeline.create(options);
            pipeline.apply("ReadLines", TextIO.read().from(options.getInputFile()))

                    .apply("Converting to TableRows", ParDo.of(new DoFn<String, TableRow>() {
                        private static final long serialVersionUID = 0;

                        @ProcessElement
                        public void processElement(ProcessContext c) {
                            String json = c.element();
                            TableRow tableRow = gson.fromJson(json, TableRow.class);
                            c.output(tableRow);
                        }
                    }))
            .apply(BigQueryIO.writeTableRows().to(options.getTableSpec())
                    .withCreateDisposition(CreateDisposition.CREATE_NEVER)
                    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

            logger.info("PIPELINE-INFO: jobName={} message={} ", jobName, "pipeline started");
            State state = pipeline.run().waitUntilFinish();
            logger.info("PIPELINE-INFO: jobName={} message={} ", jobName, "pipeline status" + state);

        } catch (Exception exception) {
            throw exception;
        }
    }
 }

Options Code:

  public interface JsonToBQTemplateOptions extends PipelineOptions{

    ValueProvider<String> getInputFile();

    void setInputFile(ValueProvider<String> value);

    ValueProvider<String> getErrorOutput();

    void setErrorOutput(ValueProvider<String> value);

    ValueProvider<String> getTableSpec();

    void setTableSpec(ValueProvider<String> value);

  }

Maven command to create template:

  mvn -X compile exec:java \
    -Dexec.mainClass=com.xyz.adp.pipeline.template.JsonTextToBqTemplate \
    -Dexec.args="--project=xxxxxx-12356 \
    --stagingLocation=gs://xxx-test/template/staging/jsontobq/ \
    --tempLocation=gs://xxx-test/temp/ \
    --templateLocation=gs://xxx-test/template/templates/jsontobq \
    --errorOutput=gs://xxx-test/template/output"

Error:

Caused by: java.lang.IllegalStateException: Cannot estimate size of a FileBasedSource with inaccessible file pattern: {}. [RuntimeValueProvider{propertyName=inputFile, default=null, value=null}]
at org.apache.beam.sdk.repackaged.com.google.common.base.Preconditions.checkState(Preconditions.java:518)
at org.apache.beam.sdk.io.FileBasedSource.getEstimatedSizeBytes(FileBasedSource.java:199)
at org.apache.beam.runners.direct.BoundedReadEvaluatorFactory$InputProvider.getInitialInputs(BoundedReadEvaluatorFactory.java:207)
at org.apache.beam.runners.direct.ReadEvaluatorFactory$InputProvider.getInitialInputs(ReadEvaluatorFactory.java:87)
at org.apache.beam.runners.direct.RootProviderRegistry.getInitialInputs(RootProviderRegistry.java:62)

Mvn Build was successful when I pass values for inputFile and tableSpec as below.

 mvn -X compile exec:java \
-Dexec.mainClass=com.ihm.adp.pipeline.template.JsonTextToBqTemplate \
-Dexec.args="--project=xxxxxx-123456 \
--stagingLocation=gs://xxx-test/template/staging/jsontobq/ \
--tempLocation=gs://xxx-test/temp/ \
--templateLocation=gs://xxx-test/template/templates/jsontobq \
--inputFile=gs://xxx-test/input/bqtest.json \
--tableSpec=xxx_test.jsontobq_test \
--errorOutput=gs://xxx-test/template/output"

But it won't create any template in Cloud dataflow.

Is there a way to create a template without validating these runtime arguments during maven execution?

Slava Chernyak Slava Chernyak · Accepted Answer · 2017-12-21T23:24:25

I think the problem here is that you are not specifying a runner. By default, this is attempting to use the DirectRunner. Try to pass

--runner=TemplatingDataflowPipelineRunner

as part of your -Dexec.args. After this you also should not need to specify the ValueProvider template arguments like inputFile, etc.

More info here:

https://cloud.google.com/dataflow/docs/templates/creating-templates

Cloud Dataflow Custom Template creation Issues

2 Answers