谷歌云数据流：CloudBigtableScanConfiguration.withScan()，如何传递动态过滤值？

Question

SourceLocation 是我的 Bigtable 的前缀，它是从 application.properties 中获取的。有没有办法在运行数据流模板时动态获取它？

我的管道：

pipeline.apply("ReadTable", Read.from(CloudBigtableIO.read(configSetUp(options))))

CloudBigtableScanConfiguration

private static CloudBigtableScanConfiguration configSetUp(LocationSetupOptions options) {
    ValueProvider<Integer>  pageFilter = options.getPageFilter();
    Scan scan = new Scan(Bytes.toBytes(options.getSourceLocation().get()));

    FilterList filterList = new FilterList();
    PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(options.getSourceLocation().get()));
    filterList.addFilter(new PageFilter(Long.valueOf(pageFilter.get())));
    filterList.addFilter(prefixFilter);
    scan.setFilter(filterList);

    return new CloudBigtableScanConfiguration.Builder()
        .withProjectId(options.getProjectId())
        .withInstanceId(options.getInstanceId())
        .withTableId(options.getTableId())
        .withScan(scan)
        .build();}

Answer 1

Bigtable CloudBigtableIO 和 BigtableIO 有两个客户端。 CloudBigtableIO 参数不会更新为通过 ValueProvider 由模板修改，但 BigtableIO 与 ValueProviders 兼容。

在您的特定情况下，如果您正在寻找与模板一起使用的ValueProvider，那么我建议您转而使用BigtableIO。可以在此处找到示例AvroToBigtable。

更新

@Default.InstanceFactory 可用于指定用户提供的工厂方法以生成参数的默认值。这样，您可以从 DefaultValueFactory 实现中的资源文件中读取默认值。

例如，您可以查看WindowedWordCount 如何定义 DefaultToCurrentSystemTime 来注释 minTimestampMillis 参数：