批之间的流数据共享火花

Question

Spark streaming以微批量处理数据。

使用RDD并行处理每个间隔数据，每个间隔之间没有任何数据共享。

但我的用例需要在间隔之间共享数据。

>

单词“hadoop”和“spark”与前一个间隔计数的相对计数

所有其他单词的正常字数。

注意：UpdateStateByKey执行有状态处理，但这将对每个记录而不是特定记录应用函数。

间隔-1

输入：

Sample Input with Hadoop and Spark on Hadoop

输出：

hadoop  2
sample  1
input   1
with    1
and 1
spark   1
on  1

Another Sample Input with Hadoop and Spark on Hadoop and another hadoop another spark spark

another 3
hadoop  1
spark   2
and 2
sample  1
input   1
with    1
on  1

火花发生3次，但输出应为2（3-1）

对于所有其他单词，它应该给出正常的字数。

因此，在处理第2个区间数据时，它应该具有第1个区间的字数hadoop和spark

Apache Storm中的类似用法：

storm中的分布式缓存

Storm TransactionalWords

Answer 1

这可以通过“记住”接收到的最后一个RDD并使用左联接将该数据与下一个流批合并来实现。我们使用streamingcontext.remember来使流式处理产生的RDD能够在我们需要的时候保留。

我们利用了以下事实：dstream.transform是在驱动程序上执行的操作，因此我们可以访问所有本地对象定义。特别是，我们希望在每个批处理上用所需的值更新到最后一个RDD的可变引用。

也许一段代码让这个想法更加清晰：

// configure the streaming context to remember the RDDs produced
// choose at least 2x the time of the streaming interval
ssc.remember(xx Seconds)  

// Initialize the "currentData" with an empty RDD of the expected type
var currentData: RDD[(String, Int)] = sparkContext.emptyRDD

// classic word count
val w1dstream = dstream.map(elem => (elem,1))    
val count = w1dstream.reduceByKey(_ + _)    

// Here's the key to make this work. Look how we update the value of the last RDD after using it. 
val diffCount = count.transform{ rdd => 
                val interestingKeys = Set("hadoop", "spark")               
                val interesting = rdd.filter{case (k,v) => interestingKeys(k)}                                
                val countDiff = rdd.leftOuterJoin(currentData).map{case (k,(v1,v2)) => (k,v1-v2.getOrElse(0))}
                currentData = interesting
                countDiff                
               }

diffCount.print()