如何保留历史 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 的相应加入键的记录时，我们执行加入但我没有不知道这段历史是如何建立起来的。

Answer 1

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

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

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

让我们看一个简单的例子，使用经典的 wordCount：

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

简而言之，txt 是 HadoopRDD(someFile)。 words 是 MapPartitionsRDD(txt, flatMapFunction)。我们将单词的lineage 称为由这种操作链接形成的 DAG（直接无环图）。：HadoopRDD <-- MapPartitionsRDD。

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

在第 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 计算的图表不断增长。 cache 降低了每次进行所有计算的成本。经常需要checkpoint 来编写这个增长图的具体计算值，以便我们可以将其用作基线，而不必评估整个链。

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

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