1

我有一个Pub/Sub主题 + 订阅，并希望在Dataflow中使用和聚合订阅中的无限数据。我使用固定窗口并将聚合写入 BigQuery。

读写（没有窗口和聚合）工作正常。但是当我将数据传输到一个固定窗口（计算每个窗口中的元素）时，该窗口永远不会被触发。因此没有写入聚合。

这是我的单词发布者（它使用示例中的 kinglear.txt作为输入文件）：

public static class AddCurrentTimestampFn extends DoFn<String, String> {
    @ProcessElement public void processElement(ProcessContext c) {
        c.outputWithTimestamp(c.element(), new Instant(System.currentTimeMillis()));
    }
}

public static class ExtractWordsFn extends DoFn<String, String> {
    @ProcessElement public void processElement(ProcessContext c) {
        String[] words = c.element().split("[^a-zA-Z']+");
        for (String word:words){ if(!word.isEmpty()){ c.output(word); }}
    }
}

// main:
Pipeline p = Pipeline.create(o); // 'o' are the pipeline options
p.apply("ReadLines", TextIO.Read.from(o.getInputFile()))
        .apply("Lines2Words", ParDo.of(new ExtractWordsFn()))
        .apply("AddTimestampFn", ParDo.of(new AddCurrentTimestampFn()))
        .apply("WriteTopic", PubsubIO.Write.topic(o.getTopic()));
p.run();

这是我的窗口字计数器：

Pipeline p = Pipeline.create(o); // 'o' are the pipeline options

BigQueryIO.Write.Bound tablePipe = BigQueryIO.Write.to(o.getTable(o))
        .withSchema(o.getSchema())
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND);

Window.Bound<String> w = Window
        .<String>into(FixedWindows.of(Duration.standardSeconds(1)));

p.apply("ReadTopic", PubsubIO.Read.subscription(o.getSubscription()))
        .apply("FixedWindow", w)
        .apply("CountWords", Count.<String>perElement())
        .apply("CreateRows", ParDo.of(new WordCountToRowFn()))
        .apply("WriteRows", tablePipe);
p.run();

上面的订阅者将不起作用，因为窗口似乎没有使用默认触发器触发。但是，如果我手动定义触发器，代码会起作用，并且计数会写入 BigQuery。

Window.Bound<String> w = Window.<String>into(FixedWindows.of(Duration.standardSeconds(1)))
        .triggering(AfterProcessingTime
                .pastFirstElementInPane()
                .plusDelayOf(Duration.standardSeconds(1)))
        .withAllowedLateness(Duration.ZERO)
        .discardingFiredPanes();

如果可能，我喜欢避免指定自定义触发器。

问题：

  1. 为什么我的解决方案不适用于 Dataflow 的默认触发器？
  2. 如何更改我的发布者或订阅者以使用默认触发器触发窗口？
4

1 回答 1

2

你如何确定触发器永远不会触发？

您的PubSubIO.Write和PubSubIO.Read转换都应使用 指定时间戳标签withTimestampLabel，否则您添加的时间戳将不会写入 PubSub 并且将使用发布时间。

无论哪种方式，管道的输入水印都将来自在 PubSub 中等待的元素的时间戳。处理完所有输入后，它将停留几分钟（以防发布者出现延迟），然后再进入实时状态。

您可能会看到所有元素都在同一个约 1 秒的窗口中发布（因为输入文件非常小）。这些都是相对较快的读取和处理，但是它们放入的 1 秒窗口要等到输入水印前进后才会触发，说明该 1 秒窗口中的所有数据都已被消耗掉。

直到几分钟后才会发生这种情况，这可能会使触发器看起来好像不起作用。您编写的触发器在 1 秒的处理时间后触发，这会更早触发，但不能保证所有数据都已处理。

从默认触发器获得更好行为的步骤：

  1. 用于withTimestampLabel写入和读取 pubsub 步骤。
  2. 让发布者进一步分散时间戳（例如，运行几分钟并将时间戳分散到该范围内）
于 2017-01-03T23:06:34.990 回答