从 Apache Spark 中的文件中联合一系列 RDD 的内存有效方法答案

【问题标题】：Memory efficient way of union a sequence of RDDs from Files in Apache Spark从 Apache Spark 中的文件中联合一系列 RDD 的内存有效方法
【发布时间】：2015-04-05 06:17:11
【问题描述】：

我目前正在尝试在 UMBC Webbase Corpus 上训练一组 Word2Vec 向量（400 个文件中大约 30GB 的文本）。

即使在 100 GB 以上的机器上，我也经常遇到内存不足的情况。我在应用程序本身中运行 Spark。我尝试稍微调整一下，但我无法对超过 10 GB 的文本数据执行此操作。我的实现的明显瓶颈是先前计算的 RDD 的联合，即内存不足异常的来源。

也许你有经验想出比这更节省内存的实现：

 object SparkJobs {
  val conf = new SparkConf()
    .setAppName("TestApp")
    .setMaster("local[*]")
    .set("spark.executor.memory", "100g")
    .set("spark.rdd.compress", "true")

  val sc = new SparkContext(conf)


  def trainBasedOnWebBaseFiles(path: String): Unit = {
    val folder: File = new File(path)

    val files: ParSeq[File] = folder.listFiles(new TxtFileFilter).toIndexedSeq.par


    var i = 0;
    val props = new Properties();
    props.setProperty("annotators", "tokenize, ssplit");
    props.setProperty("nthreads","2")
    val pipeline = new StanfordCoreNLP(props);

    //preprocess files parallel
    val training_data_raw: ParSeq[RDD[Seq[String]]] = files.map(file => {
      //preprocess line of file
      println(file.getName() +"-" + file.getTotalSpace())
      val rdd_lines: Iterator[Option[Seq[String]]] = for (line <- Source.fromFile(file,"utf-8").getLines) yield {
          //performs some preprocessing like tokenization, stop word filtering etc.
          processWebBaseLine(pipeline, line)    
      }
      val filtered_rdd_lines = rdd_lines.filter(line => line.isDefined).map(line => line.get).toList
      println(s"File $i done")
      i = i + 1
      sc.parallelize(filtered_rdd_lines).persist(StorageLevel.MEMORY_ONLY_SER)

    })

    val rdd_file =  sc.union(training_data_raw.seq)

    val starttime = System.currentTimeMillis()
    println("Start Training")
    val word2vec = new Word2Vec()

    word2vec.setVectorSize(100)
    val model: Word2VecModel = word2vec.fit(rdd_file)

    println("Training time: " + (System.currentTimeMillis() - starttime))
    ModelUtil.storeWord2VecModel(model, Config.WORD2VEC_MODEL_PATH)  
  }}
}

【问题讨论】：

文件中的 30GB 数据...肯定会导致超过 100GB 的 Java 对象...这样做以便一次只有一个文件在内存中...处理它...然后加载下一个。
另外...不要这样做 -> StorageLevel.MEMORY_ONLY_SER
我需要立即处理它们，因为在模型拟合步骤中所有数据都需要存在
我相信你可以多次训练模型。嗯......这取决于你的实施。
尝试在任何地方使用StorageLevel.MEMORY_AND_DISK...可能会有所帮助。它肯定会慢很多倍......但会起作用。

标签： scala nlp apache-spark bigdata word2vec

【解决方案1】：

您是否尝试过从较小的语料库中获取 word2vec 向量？我告诉你这是因为我在一个更小的机器上运行 word2vec spark 实现，但我遇到了问题，因为存在这个问题：http://mail-archives.apache.org/mod_mbox/spark-issues/201412.mbox/%3CJIRA.12761684.1418621192000.36769.1418759475999@Atlassian.JIRA%3E

所以对于我的用例，这个问题使 word2vec spark 实现有点没用。因此，我使用 spark 来按摩我的语料库，而不是实际获取向量。

正如其他人建议的那样，不要打电话给rdd.union。
另外我认为.toList 可能会从 RDD 中收集每一行并将其收集到您的驱动程序机器（用于提交任务的机器）中，这可能就是您内存不足的原因。您应该完全避免将 RDD 变成列表！

【讨论】：

【解决方案2】：

就像 Sarvesh 在 cmets 中指出的那样，对于单台机器来说，数据可能太多了。使用更多的机器。我们通常看到需要 20-30 GB 的内存来处理 1 GB 的文件。通过这个（非常粗略的）估计，对于 30 GB 的输入，您需要 600-800 GB 的内存。（您可以通过加载部分数据来获得更准确的估计。）

作为更一般的评论，我建议您避免使用rdd.union 和sc.parallelize。使用带有通配符的 sc.textFile 来将所有文件加载到单个 RDD 中。

【讨论】：