Apache Beam中嵌套的顶N

Question

我使用的是Apache Beam2.14和Java。

给定如下所示的数据集：

| countryID | sessionID | pageID    | count    |
| --------- | --------- | --------- | -------- |
| a         | a         | a         | 1        |
| a         | b         | c         | 2        |
| b         | c         | a         | 4        |
| c         | d         | a         | 6        |

我想返回一个数据集，它只包含计数之和位于前N个countryDs的行，对于每个countryID是前N个sessionID，对于每个sessionID是前N个pageID。

数据集的大小以数十亿行为单位--它将无法容纳在内存中。顺便说一句--数据集驻留在BigQuery中，并尝试在BigQuery中使用DENSE_RANK()或ROW_NUMBER()函数错误直接执行此操作，但由于该大小导致“内存限制超出”错误，因此尝试使用Dataflow。

我目前的策略是：

• 按countryID、sessionID、pageID的组合键分组，找出每个组的总和。
• 按countryID、sessionID对结果进行分组，并求各组之和。
• 将结果按countryID分组，并求各组之和。
• 使用top.of获取顶级countryids
• 将结果压缩回第二级分组，并使用top.perkey获取每个国家的顶级会话。
• 将结果扁平化到第一级分组，并获得每个会话的顶级页面。
• 平展结果以发出行。

棘手的部分是，需要在每个“group by”级别保留行，以便在最后发出它们。我尝试创建一个树形结构，其中每个节点保存“group by”步骤的结果--这样我就可以只计算一次它的子节点的总和，以便在后续步骤中进行比较。也就是说，在每个“group by”步骤中，结果是一个kV > ，节点具有如下字段：

    @DefaultCoder(SerializableCoder.class)
    public static class TreeNode implements Node, Serializable {
        private Long total = 0L;
        private KV<String, Iterable<LeafNode>> kv;
    ...

虽然这几乎适用于direct runner和一个小样本数据集，但在dataflow上运行时，我会遇到与节点类相关的序列化错误，因为Iterable是输入pCollection的窗口：

由:java.io.notSerializableException:org.apache.beam.runners.dataflow.worker.util.BatchGroupSobyWindowviaiteratorsFn$Windowreterable

（参见https://beam.apache.org/releases/javadoc/2.15.0/index.html？org/apache/beam/sdk/transforms/groupbykey.html）

考虑到我需要处理的数据集大小，将数据复制到内存中的不同集合以实现可序列化不是一个可行的选择。

以下是到目前为止管道的一个示例--仅使用2个分组级别作为示例：

        Pipeline pipeline = Pipeline.create(options);

        pipeline.apply("Read from BQ", BigQueryIO.readTableRows().from(inputTable))
                .apply("Transform to row", ParDo.of(new RowParDo())).setRowSchema(SCHEMA)
                .apply("Set first level key", WithKeys.of(new GroupKey(key1)))
                .apply("Group by", GroupByKey.create())
                .apply("to leaf nodes", ParDo.of(new ToLeafNode()))
                .apply("Set 2nd level key", WithKeys.of(new GroupKey2()))
                .apply("Group by 2nd level", GroupByKey.create())
                .apply("To tree nodes", ParDo.of(new ToTreeNode()))
                .apply("Top N", Top.of(10, new CompareTreeNode<TreeNode>()))
                .apply("Flatten", FlatMapElements.via(new FlattenNodes<TreeNode>()))
                .apply("Expand", ParDo.of(new ExpandTreeNode()))
                .apply("Top N of first key", Top.perKey(10, new CompareTreeNode<LeafNode>()))
                .apply("Values", Values.create())
                .apply("Flatten", FlatMapElements.via(new FlattenNodes<LeafNode>()))
                .apply("Expand", ParDo.of(new ExpandLeafNode()))
                .apply("Values", Values.create())
                .apply("Write to bq",
                        BigQueryIO.<Row>write().to(outputTable).withSchema(BigQueryUtils.toTableSchema(SCHEMA))
                                .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                                .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                                .withFormatFunction(BigQueryUtils.toTableRow()));
        pipeline.run();

似乎这应该是一个共同的目标，所以我想知道是否有更简单的方法，或者任何使用Beam在Java中实现同样事情的例子。

Answer 1

您可以尝试使用setcoder设置代码，如下所示。

    Pipeline pipeline = Pipeline.create(options);

    pipeline.apply("Read from BQ", BigQueryIO.readTableRows().from(inputTable))
            .apply("Transform to row", ParDo.of(new RowParDo())).setRowSchema(SCHEMA)
            .apply("Set first level key", WithKeys.of(new GroupKey(key1)))
            .apply("Group by", GroupByKey.create())
            .apply("to leaf nodes", ParDo.of(new ToLeafNode()))
            .apply("Set 2nd level key", WithKeys.of(new GroupKey2()))
            .apply("Group by 2nd level", GroupByKey.create())
            .apply("To tree nodes", ParDo.of(new ToTreeNode())).setCoder(SerializableCoder.of(TreeNode.class))
            .apply("Top N", Top.of(10, new CompareTreeNode<TreeNode>()))
            .apply("Flatten", FlatMapElements.via(new FlattenNodes<TreeNode>()))
            .apply("Expand", ParDo.of(new ExpandTreeNode()))
            .apply("Top N of first key", Top.perKey(10, new CompareTreeNode<LeafNode>()))
            .apply("Values", Values.create())
            .apply("Flatten", FlatMapElements.via(new FlattenNodes<LeafNode>()))
            .apply("Expand", ParDo.of(new ExpandLeafNode()))
            .apply("Values", Values.create())
            .apply("Write to bq",
                    BigQueryIO.<Row>write().to(outputTable).withSchema(BigQueryUtils.toTableSchema(SCHEMA))
                            .withWriteDisposition(WriteDisposition.WRITE_TRUNCATE)
                            .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
                            .withFormatFunction(BigQueryUtils.toTableRow()));
    pipeline.run();

但是，对于需要确定前N个国家、前N个会话和前N个页面的情况，我建议将管道简化为只对右侧字段单独分组，然后应用sum和Top，如下所示。

    Pipeline pipeline = Pipeline.create(options);

    rows = pipeline.apply("Read from BQ", BigQueryIO.readTableRows().from(inputTable))
           .apply("Transform to row", ParDo.of(new RowParDo())).setRowSchema(SCHEMA);
    sumByCountry =rows.apply("Set Country key", MapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers()))
         .via((Row row) -> KV.of(row.getCountry(), row.getCount()))))
         .apply("Country Scores", Sum.<String>integersPerKey());
         .apply("Top Countries", Top.of(N, new CompareValues()))
     // Do the same for Session and page
     sumBySession = rows....
     sumByPage = rows....

我不确定您是否想要获得前N个国家的所有行，但如果您想要获得行，那么您可以在行PCollection上使用前N个国家的侧输入并过滤出结果。您可以对会话和页面执行相同的操作。

Dataflow应该根据这个用例的需要进行伸缩，所以您不需要为这个用例手动执行中间groupby。