在 Dataflow 中执行聚合

Question

我将（时间序列）值存储在 Bigtable 中，我遇到了一个用例，我需要在这些值上应用 filter 并执行聚合。我正在使用以下配置来获取connection 到Bigtable（执行范围扫描等）：

Connection connection = BigtableConfiguration.connect(projectId, instanceId);
Table table = connection.getTable(TableName.valueOf(tableId)); 

table.getScanner(<a scanner with filter>);

这对ResultScanner 有帮助，我可以迭代行。但是，我想要做的是对某些列执行聚合并获取值。我想要做的一个 SQL 等价物是这样的：

SELECT SUM(A), SUM(B)
FROM table
WHERE C = D;

为了在HBase 中做同样的事情，我遇到了AggregationClient (javadoc here)，但是，它需要Configuration 并且我需要在Bigtable 之外运行的东西（这样我就不需要使用低级 Hbase API）。

我检查了文档，但找不到任何可以做到这一点的（Java 语言）。任何人都可以分享一个示例以在 BigTable 上使用（非行键或任何）过滤器执行aggregation。

Answer 1

Bigtable 本身没有任何聚合机制。另外，Bigtable 处理WHERE C = D 有困难，所以这种类型的处理一般最好在客户端完成。

AggregationClient 是一个 HBase 协处理器。 Cloud Bigtable 不支持协处理器。

如果您想将 Cloud Bigtable 用于此类聚合，则必须使用 table.scan() 和您自己的逻辑。如果规模足够大，您将不得不使用 Dataflow 或 BigQuery 来执行聚合。

Answer 2

这是一种方法：

PCollection<TableRow> rows = p.apply(BigQueryIO.readTableRows()
  .fromQuery("SELECT A, B FROM table;"));

PCollection<KV<String, Integer>> valuesA =
  rows.apply(
    MapElements.into(TypeDescriptors.kvs(
      TypeDescriptors.strings(),
      TypeDescriptors.integers()))
      .via((TableRow row) -> KV.of(
        "A", (Integer) row.getF().get(0).getV())));

PCollection<KV<String, Integer>> valuesB =
  rows.apply(
    MapElements.into(TypeDescriptors.kvs(
      TypeDescriptors.strings(),
      TypeDescriptors.integers()))
      .via((TableRow row) -> KV.of(
        "B", (Integer) row.getF().get(1).getV())));

PCollection<KV<String, Integer>> sums =
  PCollectionList.of(sumOfA).and(sumOfB)
    .apply(Flatten.pCollections())
    .apply(Sum.integersPerKey());