如何计算每个窗口的元素 [英] How to count elements per window
问题描述
我正在尝试解决一个看似简单的问题——计算每个窗口的 PCollection 中有多少元素.我需要它在写入时传递给 .withSharding() 函数,以创建与要写入的文件一样多的分片.
I'm trying to solve what seems to be easy problem -- count how many elements there are in a PCollection per window. I need it to pass to .withSharding() function on write, to create as many shards as there are going to be files to write.
我尝试这样做:
FileIO.writeDynamic<Long, E>()
.withDestinationCoder(AvroCoder.of(Long::class.java))
.by { e -> e.key }
.via(Contextful.fn(MySerFunction()))
.withNaming({ key -> MyFileNaming() })
.withSharding(ShardingFn())
.to("gs://some-output")
class ShardingFn : PTransform<PCollection<E>>, PCollectionView<Int>>() {
override fun expand(input: PCollection<E>): PCollectionView<Int> {
val keys: PCollection<Long> = input.apply(Keys.create())
// This only works with GlobalWindowing, how to count per window?
val count: PCollection<Long> = keys.apply(Count.globally())
val int: PCollection<Int> = count.apply(MapElements.via(Long2Int))
return int.apply(View.asSingleton())
}
但是,这仅在我有全局窗口(又名批处理模式")时才有效,否则 Count.globally() 将抛出异常.
However, this works only as long as I have global windowing (aka "batch mode"), otherwise Count.globally() will throw an exception.
也许我写错了,但如果我因为其他原因想计算每个窗口的元素,该怎么做?
Maybe I'm doing it wrong for writing, but if I ever want to count elements per window for some other reason, how to do that?
推荐答案
使用 Combine.globally(Count.<T>combineFn()).withoutDefaults()
而不是 Count.global()
应该适用于您的情况.这也可以在 Javadoc 中找到:https://beam.apache.org/documentation/sdks/javadoc/2.5.0/org/apache/beam/sdk/transforms/Count.html#globally--
Using Combine.globally(Count.<T>combineFn()).withoutDefaults()
instead of Count.globally()
should work in your case. This can also be found in the Javadoc: https://beam.apache.org/documentation/sdks/javadoc/2.5.0/org/apache/beam/sdk/transforms/Count.html#globally--
这篇关于如何计算每个窗口的元素的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!