相关疑难解决方法(0)

我试图使用本指南在EC2上使用Spark主机对常见爬网数据进行简单转换,我的代码如下所示:

package ccminer

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

object ccminer {
  val english = "english|en|eng"
  val spanish = "es|esp|spa|spanish|espanol"
  val turkish = "turkish|tr|tur|turc"
  val greek = "greek|el|ell"
  val italian = "italian|it|ita|italien"
  val all = (english :: spanish :: turkish :: greek :: italian :: Nil).mkString("|")

  def langIndep(s: String) = s.toLowerCase().replaceAll(all, "*")

  def main(args: Array[String]): Unit = {
    if (args.length != 3) {
      System.err.println("Bad command line")
      System.exit(-1)
    }

    val cluster = "spark://???"
    val sc = new SparkContext(cluster, "Common Crawl Miner", …

在Spark中,每次我们对RDD执行任何操作时,都会重新计算RDD.因此,如果我们知道RDD将被重用,我们应该明确地缓存RDD.

让我们说,Spark决定懒惰地缓存所有RDD并使用LRU自动将最相关的RDD保存在内存中(这是大多数缓存以任何方式工作的方式).这对开发人员有很大的帮助,因为他不必考虑缓存并专注于应用程序.此外,我不知道它如何对性能产生负面影响,因为很难跟踪程序内部使用变量(RDD)的次数,大多数程序员将决定以任何方式缓存大多数RDD.

缓存通常会自动发生.以OS /平台或框架或工具为例.但是由于分布式计算中缓存的复杂性,我可能会忽略为什么缓存不能自动或性能影响.

所以我无法理解,为什么我必须显式缓存为,

1. 它看起来很难看
2. 它很容易被遗漏
3. 它可以很容易地使用过度/不足

我在Youtube上看过一些关于Spark架构的视频.

即使Lazy评估,在出现故障时数据创建的弹性,良好的函数式编程概念是Resilenace分布式数据集成功的原因,但令人担忧的一个因素是由于多次转换导致的内存开销导致数据不可变导致的内存开销.

如果我正确地理解了这个概念,那么每个转换都会创建新的数据集,因此内存需求将会多次消失.如果我在代码中使用10个转换,将创建10组数据集,并且我的内存消耗将增加10倍.

例如

val textFile = sc.textFile("hdfs://...")
val counts = textFile.flatMap(line => line.split(" "))
                 .map(word => (word, 1))
                 .reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://...")

以上示例有三个转换:flatMap, map and reduceByKey.这是否意味着对于X大小的数据我需要3X数据内存？

我的理解是否正确？缓存RDD只是解决此问题的解决方案吗？

一旦我开始缓存,它可能会溢出到磁盘,因为大尺寸和性能会因磁盘IO操作而受到影响.在这种情况下,Hadoop和Spark的性能是否相当？

编辑:

从答案和评论中,我已经理解了延迟初始化和管道流程.我假设3 X内存,其中X是初始RDD大小不准确.

但是有可能在内存中缓存1 X RDD并通过pipleline更新它吗？cache()如何工作？