python之Spark Streaming之计算状态中的不同元素

我有一个带有 VideoID-UserID 键值对的 dstream，按 VideoID 计算不同的 UserID 组的好做法是什么？

// VideoID,UserID 
foo,1 
foo,2 
bar,1 
bar,2 
foo,1 
bar,2 

如上，我想随时去掉多余的foo,1和bar,2得到VideoID-CountUserID，所以结果应该是:

foo: 2 
bar: 2 

换句话说，我想在内存中保存一个大的状态数据集。当一批新的 dstream 到达时，将其与数据集进行比较以计算每个视频的不同用户。

怎么做？

我正在研究 Spark 1.6，但接受了更高版本的答案。 Python 代码(如果可能)。

请您参考如下方法:

为了获得按视频 ID 分组的用户 ID 的不同计数，请考虑使用 aggregateByKey。抱歉，这是 Scala，所以你必须翻译。

val rdd = sc.textFile("your_file.txt") 
 
val initialSet = Set.empty[Int] 
val addToSet = (s: Set[Int], v:Int) => s + v 
val mergeSets = (s1: Set[Int], s2: Set[Int]) => s1 ++ s2 
 
val distinctValSets = rdd.aggregateByKey(initialSet)(addToSet, mergeSets) 
val distinctValCountd = rdd.map({case(k,s) => (k,s.size)}) 

Initial set 是你的聚合对象的初始值，addToSet 和mergeSets 指定如何将值添加到你的集合中并根据键合并不同的集合。这应该为您提供与每个视频关联的不同数量的用户，并且比 reduceByKey 和 groupByKey 更便宜(空间方面)。