如何保存历史RDD以在给定代码中进一步使用

Question

{
var history: RDD[(String, List[String]) = sc.emptyRDD()

val dstream1 = ...
val dstream2 = ...

val historyDStream = dstream1.transform(rdd => rdd.union(history))
val joined = historyDStream.join(dstream2)

... do stuff with joined as above, obtain dstreamFiltered ...

dstreamFiltered.foreachRDD{rdd =>
val formatted = rdd.map{case (k,(v1,v2)) => (k,v1) }
history.unpersist(false) // unpersist the 'old' history RDD
history = formatted // assign the new history
history.persist(StorageLevel.MEMORY_AND_DISK) // cache the computation
history.count() //action to materialize this transformation
}

这个代码逻辑可以很好地保存所有以前的RDD，这些RDD没有成功地加入并保存到未来的批处理中，因此每当我们获得一个带有该RDD相应加入键的记录时，我们都会执行加入，但我不知道这段历史是如何建立的。

Answer 1

我们可以通过观察RDD谱系是如何随时间演化的，来理解在这种情况下历史是如何建立的。

我们需要两条先前的知识：

• RDD是不可变结构
• RDD上的操作可以通过要应用的函数和对输入RDD的引用以功能术语表示。

让我们看一个使用经典字数的快速示例：

val txt = sparkContext.textFile(someFile)
val words = txt.flatMap(_.split(" "))

简而言之，txt是一个HadoopRDD（someFile）words（代码>单词）是一个映射分区（txt，flatMapFunction）。我们把单词的谱系称为DAG（直接无环图），它是由操作链构成的<代码>HadoopRDD

我们可以将相同的原则应用于流媒体操作：

在迭代0中，我们有

var history: RDD[(String, List[String]) = sc.emptyRDD()
// -> history: EmptyRDD
...
val historyDStream = dstream1.transform(rdd => rdd.union(history))
// -> underlying RDD: rdd.union(EmptyRDD)
join, filter
// underlying RDD: ((rdd.union(EmptyRDD).join(otherRDD)).filter(pred)
map
// -> underlying RDD: ((rdd.union(EmptyRDD).join(otherRDD)).filter(pred).map(f) 
history.unpersist(false)
//  EmptyRDD.unpersist (does nothing, it was never persisted)
history = formatted
// history =  ((rdd.union(EmptyRDD).join(otherRDD)).filter(pred).map(f)
history.persist(...)
// history marked for persistence (at the next action)
history.count()
// ((rdd.union(EmptyRDD).join(otherRDD)).filter(pred).map(f).count()
// cache result of:  ((rdd.union(EmptyRDD).join(otherRDD)).filter(pred).map(f)

在迭代1中，我们有（添加rdd0、rdd1作为迭代索引）：

val historyDStream = dstream1.transform(rdd => rdd.union(history))
// -> underlying RDD: rdd1.union(((rdd0.union(EmptyRDD).join(otherRDD0)).filter(pred).map(f))
join, filter
// underlying RDD: ((rdd1.union(((rdd0.union(EmptyRDD).join(otherRDD0)).filter(pred).map(f)).join(otherRDD1)).filter(pred)
map
// -> underlying RDD: ((rdd1.union(((rdd0.union(EmptyRDD).join(otherRDD0)).filter(pred).map(f)).join(otherRDD1)).filter(pred).map(f) 
history.unpersist(false)
//  history0.unpersist (marks the previous result for removal, we used it already for our computation above)
history = formatted
// history1 =  ((rdd1.union(((rdd0.union(EmptyRDD).join(otherRDD0)).filter(pred).map(f)).join(otherRDD1)).filter(pred).map(f)
history.persist(...)
// new history marked for persistence (at the next action)
history.count()
// ((rdd1.union(((rdd0.union(EmptyRDD).join(otherRDD0)).filter(pred).map(f)).join(otherRDD1)).filter(pred).map(f).count()
// cache result sothat we don't need to compute it next time  

这个迭代过程会随着每次迭代而继续。

正如我们所看到的，表示RDD计算的图在不断增长<代码>缓存减少了每次进行所有计算的成本<代码>检查点（code>checkpoint）经常需要为这个不断增长的图编写一个具体的计算值，以便我们可以使用它作为基线，而不必评估整个链。

查看此过程的一种有趣方式是在foreachRDD中添加一行来检查当前沿袭：

...
history.unpersist(false) // unpersist the 'old' history RDD
history = formatted // assign the new history
println(history.toDebugString())
...