apache-spark - Spark Streaming - DStream 没有 distinct()

Question

我想计算一些表示为 RDD 的 ID 的不同值。

在非流媒体的情况下，它相当简单。假设 IDs 是从平面文件中读取的 ID 的 RDD。

    print ("number of unique IDs %d" %  (IDs.distinct().count()))

但我似乎无法在流媒体案例中做同样的事情。假设我们有 streamIDs 是从网络读取的 ID 的 DStream。

    print ("number of unique IDs from stream %d" %  (streamIDs.distinct().count()))

给我这个错误

AttributeError: 'TransformedDStream' object has no attribute 'distinct'

我做错了什么？如何打印此批处理中出现的不同 ID 的数量？

Answer 1

使用 RDD，您有一个结果，但使用 DStream，您有一系列结果，每个微批处理都有一个结果。因此，您不能一次打印唯一 ID 的数量，而是必须注册一个操作来打印每个微批处理的唯一 ID，这是一个可以在其上使用 distinct 的 RDD:

streamIDs.foreachRDD(rdd => println(rdd.distinct().count()))

请记住，您可以使用 window 来创建具有更大批处理的转换后的 dstream:

streamIDs.window(Duration(1000)).foreachRDD(rdd => println(rdd.distinct().count()))