dstream中的独特元素

1 回答

• 2

  我们可以使用一个t 4间隔的窗口来保持“最近最近看到的密钥”的计数，并使用它来删除当前间隔的重复项 .

  有点像这样：

  // original dstream
  val dstream = ??? 
  // make distinct (for a single interval) and pair with 1's for counting
  val keyedDstream = dstream.transform(rdd=> rdd.distinct).map(e => (e,1))
  // keep a window of t*4 with the count of distinct keys we have seen
  val windowed = keyedDstream.reduceByKeyAndWindow((x:Int,y:Int) => x+y, Seconds(4),Seconds(1))
  // join the windowed count with the initially keyed dstream
  val joined = keyedDstream.join(windowed)
  // the unique keys though the window are those with a running count of 1 (only seen in the current interval) 
  val uniquesThroughWindow = joined.transform{rdd => 
      rdd.collect{case (k,(current, prev)) if (prev == 1) => k}
  }
  

  回复于 2024-05-06T06:59:36+08:00