Using run time parameters with BigtableIO in Apache Beam

Question

I am trying to use run time parameters with BigtableIO in Apache Beam to write to BigTable.

I have created a pipeline to read from BigQuery and writing to Bigtable. The pipeline works fine when i provide static parameters (using ConfigBigtableIO and ConfigBigtableConfiguration, referring to example here - https://github.com/GoogleCloudPlatform/cloud-bigtable-examples/blob/master/java/dataflow-connector-examples/src/main/java/com/google/cloud/bigtable/dataflow/example/HelloWorldWrite.java) but I am getting a compile error while trying to setup the pipeline with run time parameters. The options is setup with all parameters being runtime Value Providers.

    p.apply(BigQueryIO.readTableRows().fromQuery(options.getBqQuery())
            .usingStandardSql())
            .apply(ParDo.of(new TransFormFn(options.getColumnFamily(), options.getRowKey(), options.getColumnKey(), options.getRowKeySuffix())))

            .apply(BigtableIO.write().withProjectId(options.getBigtableProjectId()).
                    withInstanceId(options.getBigtableInstanceId()).
                   withTableId(options.getBigtableTableId()));

It is expecting the output of Bigtable.write()... to be org.apache.beam.sdk.transforms.PTransform,OutputT> while Bigtable.write() is returning a Write object. Can you help with providing the correct syntax to fix this? Thanks.

Here's the definition of the Write class: github.com/apache/beam/blob/release-2.4.0/sdks/java/io/…: public abstract static class Write extends PTransform<PCollection<KV<ByteString, Iterable<Mutation>>>, PDone> — Solomon Duskis

Kevin Si Kevin Si · Accepted Answer · 2018-05-09T12:46:22

Runtime parameters are meant to be used in Dataflow templates.

Are you trying to create a template and run the pipeline using the template? If yes, you would need following steps:

Create an Options that has runtime parameters you need, i.e.
ValueProvider tableId.
Pass these runtime parameters to the config object: i.e. withTableId(ValueProvider tableId) =>
withTableId(options.getTableId())
Construct your template
Execute your pipeline using the template.

The advantage of using a template is that it allows pipeline to be constructed once and executed multiple times later with runtime parameters. For more information on how to use Dataflow template: https://cloud.google.com/dataflow/docs/templates/overview

When not using Dataflow template, you don't have set runtime parameters, i.e. withTableId(ValueProvider tableId). Instead, use withTableId(String tableId).

Hope this helps!

Using run time parameters with BigtableIO in Apache Beam

1 Answers