跨 dstream 的不同元素

Posted 2023-04-15

【中文标题】跨 dstream 的不同元素

【英文标题】：Distinct Element across dstream

【发布时间】：2016-06-07 17:16:42

【问题描述】：

我正在研究窗口 dstream，其中每个 dstream 包含 3 个带有以下键的 rdd：

a,b,c
b,c,d
c,d,e
d,e,f

我只想获取所有 dstream 中的唯一键

a,b,c,d,e,f

如何在火花流中做到这一点？

【问题讨论】：

仅适用于 (3-4) 个间隔的每个窗口？还是您想要所有见过的唯一键？

你好 maasg，我的应用程序不会在 4 分钟的窗口之后发送重复。因此，如果“a”出现在时间“t”，则它有可能出现在“t+1”、“t+2”、“t+3”和“t+4”而不超出此时间。感谢您考虑解决方案，

我尝试创建具有一定大小和滑动窗口的 dstream 窗口。和 reduceByKey 。它有助于以时间为代价减少一定比例的重复。例如： Window(2mins,2mins) :- 将生成 dstream1 RDD - (rdd1: a,b,c ; rdd2:b,c,d) 并且 reduceByKey 将输出 a,b,c,d 但下一个dstream2 RDD - rdd1:c,d,e ; rdd2:d,e,f 将输出 c,d,e,f。 dstream1 和 dstream2 之间有重复 (c,d) .. 我的应用程序延迟在这里增长了 2 分钟，其中一些重复减少了。这不是正确的做法。

【参考方案1】：

我们可以使用 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.transformrdd => 
    rdd.collectcase (k,(current, prev)) if (prev == 1) => k

【讨论】：

有效！谢谢@maasg！！！但是，reduceByKeyAndWindow 没有给出，并且“2”的价值不知道为什么。因此，而不是 val windowed = keyedDstream.reduceByKeyAndWindow((x:Int,y:Int) => x+y, Seconds(4),Seconds(1)) 我不得不做 val windowed = keyedDstream.window(Seconds(4 ),Seconds(1).reduceByKey((x:Int,y:Int) => x+y )).

Probabay reduceByKeyAndWindow 仍然无法在 dstreams 上工作 ***.com/questions/29961925/…

您使用的是什么版本的 Spark？我实际上在发布之前测试了这种方法。这比预期的更具挑战性。

Spark 版本 1.6.0。感谢您抽出宝贵时间