在 Apache Beam 中通过 BigtableIO 使用运行时参数答案

【问题标题】：Using run time parameters with BigtableIO in Apache Beam在 Apache Beam 中通过 BigtableIO 使用运行时参数
【发布时间】：2018-10-02 08:26:48
【问题描述】：

我正在尝试在 Apache Beam 中使用运行时参数和 BigtableIO 来写入 BigTable。

我创建了一个管道来读取 BigQuery 并写入 Bigtable。当我提供静态参数（使用 ConfigBigtableIO 和 ConfigBigtableConfiguration，参考此处的示例 - https://github.com/GoogleCloudPlatform/cloud-bigtable-examples/blob/master/java/dataflow-connector-examples/src/main/java/com/google/cloud/bigtable/dataflow/example/HelloWorldWrite.java）时，管道工作正常，但在尝试使用运行时参数设置管道时出现编译错误。选项设置为所有参数都是运行时值提供者。

    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()));

期望 Bigtable.write()... 的输出为 org.apache.beam.sdk.transforms.PTransform,OutputT> 而 Bigtable.write() 正在返回一个 Write 对象。你能帮助提供正确的语法来解决这个问题吗？谢谢。

【问题讨论】：

Write 类的定义如下：github.com/apache/beam/blob/release-2.4.0/sdks/java/io/…: public abstract static class Write extends PTransform>>, PDone>

标签： java google-cloud-dataflow apache-beam google-cloud-bigtable

【解决方案1】：

运行时参数旨在用于 Dataflow 模板。

您是否尝试创建模板并使用该模板运行管道？如果是，您需要以下步骤：

创建一个包含您需要的运行时参数的选项，即
ValueProvider tableId。
将这些运行时参数传递给配置对象：即 withTableId(ValueProvider tableId) =>
withTableId(options.getTableId())
构建您的模板
使用模板执行您的管道。

使用模板的优点是它允许管道构建一次，然后使用运行时参数执行多次。有关如何使用 Dataflow 模板的更多信息：https://cloud.google.com/dataflow/docs/templates/overview

不使用 Dataflow 模板时，您没有设置运行时参数，即 withTableId(ValueProvider tableId)。而是使用 withTableId(String tableId)。

希望这会有所帮助！

【讨论】：