在固定窗口内对元素进行排序 - Cloud Dataflow答案

【问题标题】：Sort elements within a fixed window - Cloud Dataflow在固定窗口内对元素进行排序 - Cloud Dataflow
【发布时间】：2019-05-03 11:11:42
【问题描述】：

我有一个数据流管道，它从 pubsub 主题读取、执行转换并写入 BigTable。我希望从 pubsub 读取的元素按照它们的序列号进行处理。

我使用 2 分钟的固定窗口，然后在其上应用 GroupByKey。在 GBK 之后，我使用了一个 SortValues 转换，它对 SequenceNumber 上的 Iterable 进行排序。我观察到 GroupByKey 步骤的挂墙时间很高，因为窗口中的所有元素都在同一个工作人员上处理。有没有一种有效的方法对固定窗口中的元素进行排序？

以下是我的管道代码：

PCollection<PubsubMessage> pubsubRecords = p.apply(PubsubIO.readMessagesWithAttributes()
                    .fromTopic(StaticValueProvider.of(topic)));
            PCollection<KV<BigInteger, JSONObject>> window = pubsubRecords.apply("Raw to String", ParDo.of(new LogsFn()))
                    .apply("Window", Window
                            .<KV<BigInteger, JSONObject>>into(FixedWindows.of(Duration.standardMinutes(2)))
                            .triggering(Repeatedly
                                .forever(AfterProcessingTime
                                    .pastFirstElementInPane()
                                    .plusDelayOf(Duration.StandardMinutes(2))
                                )
                            )
                            .withAllowedLateness(Duration.ZERO).discardingFiredPanes()
                        );
            PCollection<KV<String, KV<BigInteger, JSONObject>>> keyedWindow = window
                    .apply(WithKeys.of(new SerializableFunction<KV<BigInteger, JSONObject>,String>() {
                          @Override
                          public String apply(KV<BigInteger, JSONObject> row) {
                            return "key";
                          }
                    }));

            PCollection<KV<String, Iterable<KV<BigInteger, JSONObject>>>> groupedWindow = keyedWindow
                    .apply(GroupByKey.<String, KV<BigInteger, JSONObject>>create()).apply(
                            SortValues.<String, BigInteger, JSONObject>create(BufferedExternalSorter.options()));

【问题讨论】：

您是否有理由对所有值进行排序而不是获取一些最大/最小值？需要有序元素的用例是什么？
我正在从 dynamoDB 复制数据库操作。我正在将 dynamoDB 流读入 pubsub 主题。事件必须排序，因为它们都是数据库操作

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

【解决方案1】：

我认为你的方法是正确的。不可避免的是，您应该在同一个工作人员中对所有元素进行排序。有序处理会在数据之间产生依赖关系，并且通常不适用于分布式计算。

【讨论】：

次要编辑：每个键的所有元素都将位于该窗口的同一个工作器上。