Count distinct values in a stream pipeline

Question

I have a pipeline that looks like

pipeline.apply(PubsubIO.read.subscription("some subscription"))
            .apply(Window.into(SlidingWindow.of(10 mins).every(20 seconds)
                            .triggering(AfterProcessingTime.pastFirstElementInPane()
                    .plusDelayOf(20 seconds))
                    .withAllowedLateness(Duration.ZERO)
                    .accumulatingFiredPanes()))
            .apply(RemoveDuplicates.create())
            .apply(Window.discardingFiredPanes()) // this is suggested in the warnings under https://cloud.google.com/dataflow/model/triggers#window-accumulation-modes
            .apply(Count.<String>globally().withoutDefaults())

This pipeline overcounts distinct values significantly (20x normal value). Initially, I was suspecting that the default trigger may have caused this issue. I have tweaked to use triggers that allow no lateness/discard fired panes/use processing time, all of which have similar overcount issues.

I've also tried ApproximateUnique.globally: it failed during pipeline construction because of an exception that looks like Default values are not supported in Combine.globally() if the output PCollection is not windowed by GlobalWindows. There seems no way to add withoutDefaults to it (like we did with Count.globally).

Is there a recommended way to do COUNT(DISTINCT) in dataflow/beam streaming pipeline with reasonable precision?

P.S. I'm using Java Dataflow SDK 1.9.0.

Answer 1

Your code looks OK; it shouldn't overcount. Note that you are placing each element into 30 windows, so if you have a window-unaware sink (equivalent to collapsing all the sliding windows) you would expect precisely 30 times as many elements. If you could show a bit more of the pipeline or how you are observing the counts, that might help.

Beyond that, I have a few suggestions for the pipeline:

• I suggest changing your trigger for RemoveDuplicates to AfterPane.elementCountAtLeast(1); this will get you the same result at lower latency, since later elements arriving will have no impact. This trigger, and your current trigger, will never fire repeatedly. So it does not actually matter whether you set accumulatingFiredPanes() or discardingFiredPanes(). This is good, because neither one would work with the rest of your pipeline.
• I'd install a new trigger prior to the Count. The reason is a bit technical, but I'll try to describe it:
  • In your current pipeline, the trigger installed there (the "continuation trigger" of the trigger for RemoveDuplicates) notes the arrival time of the first element and waits until it has received all elements that were produced at or before that processing time, as measured by the upstream worker. There is some nondeterminism because it puns the local processing time and the processing time of other workers.
  • If you take my advice and switch the trigger for RemoveDuplicates, then the continuation trigger will be AfterPane.elementCountAtLeast(1) so it will always emit a count as soon as possible and then discard further data, which is very wrong.