可以在 apache Beam 中为 bigtable 编写动态行键过滤器答案

【问题标题】：It's posible write a dynamic rowkey filter in apache beam for bigtable可以在 apache Beam 中为 bigtable 编写动态行键过滤器
【发布时间】：2020-09-26 02:35:14
【问题描述】：

我正在寻找一种基于数据流作业中的当前时间进行过滤的方法。

我有以下代码要从 bigtable 读取，但我找不到生成动态扫描器或可调用函数或 lamda 的方法，以便能够将日期作为搜索参数传递。

RowFilter filter = RowFilter.newBuilder().setRowKeyRegexFilter(ByteString.copyFromUtf8("2020-06-05#.*#")).build(); 
PCollection<ObjectDto> collection = pipeline.apply("Read",
        BigtableIO.read()
                .withBigtableOptions(optionsBuilder)
                .withTableId("table")
                .withRowFilter(filter));

我需要这样的东西

PCollection<ObjectDto> collection = pipeline.apply("Read",
        BigtableIO.read()
                .withBigtableOptions(optionsBuilder)
                .withTableId("table")
                .withRowFilter(RowFilter.newBuilder().setRowKeyRegexFilter(
                    new SerializableFunction(){
                        Date d = new Date();
                        return DateFormat.getDate(d) + "#.*#"+DateFormat.getTime(d);
                    }  
                ));

【问题讨论】：

标签： google-cloud-platform google-cloud-dataflow google-cloud-bigtable apache-beam-io

【解决方案1】：

请注意，目前BigTableIO.Read 类仅支持ValueProvider 指定ProjectId、InstanceId 或TableId。这允许创建模板并动态更改这些参数；但是，目前无法以这种方式传递RowFilter 对象。

但是，如果您在没有模板的情况下运行管道，则可以为您的用例创建 Date RowFilter，如下所示：

DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
Date date = new Date();
String currentDate = dateFormat.format(date) + "#.*#";
RowFilter filter = RowFilter.newBuilder().setRowKeyRegexFilter(ByteString.copyFromUtf8(currentDate)).build();

然后，将其传递给您的管道：

pipeline.apply("Read",
   BigtableIO.read()
      .withProjectId("<PROJECT_ID>")
       .withInstanceId("<INSTANCE_ID>")
       .withTableId(“TABLE_ID”)
       .withRowFilter(filter));

这样，每次运行管道时，都会使用当前日期创建过滤器。

【讨论】：