google-cloud-dataflow - 如何在 Dataflow 中创建用户定义的计数器？

Question

如何在我的 DoFns 中创建自己的计数器？

在我的 DoFn 中，我想在处理记录时每次满足条件时增加一个计数器。我希望这个计数器对所有记录的值求和。

Answer 1

您可以使用Aggregators，计数器的总值将显示在 UI 中。

这是一个示例，我在管道中试验了聚合器，该管道仅使 numOutputShards 工作人员睡眠 sleepSecs 秒。（开头的 GenFakeInput PTransform 只返回一个扁平的 PCollection<String> 大小为 numOutputShards）：

PCollection<String> output = p
    .apply(new GenFakeInput(options.getNumOutputShards()))
    .apply(ParDo.named("Sleep").of(new DoFn<String, String>() {
         private Aggregator<Long> tSleepSecs;
         private Aggregator<Integer> tWorkers;
         private Aggregator<Long> tExecTime;
         private long startTimeMillis;

         @Override
         public void startBundle(Context c) {
           tSleepSecs = c.createAggregator("Total Slept (sec)", new Sum.SumLongFn());
           tWorkers = c.createAggregator("Num Workers", new Sum.SumIntegerFn());
           tExecTime = c.createAggregator("Total Wallclock (sec)", new Sum.SumLongFn());
           startTimeMillis = System.currentTimeMillis();
         }

         @Override
         public void finishBundle(Context c) {
           tExecTime.addValue((System.currentTimeMillis() - startTimeMillis)/1000);
         }

         @Override
         public void processElement(ProcessContext c) {
           try {
             LOG.info("Sleeping for {} seconds.", sleepSecs);
             tSleepSecs.addValue(sleepSecs);
             tWorkers.addValue(1);
             TimeUnit.SECONDS.sleep(sleepSecs);
           } catch (InterruptedException e) {
             LOG.info("Ignoring caught InterruptedException during sleep.");
           }
           c.output(c.element());
         }}));