apache-spark - 使用 text8 文件的 Spark Word2Vec 示例

Question

我正在尝试使用他们在其网站上引用的 text8 文件( https://spark.apache.org/docs/latest/mllib-feature-extraction.html )从 apache.spark.org 运行此示例(代码如下，整个教程在这里: http://mattmahoney.net/dc/text8.zip ):

import org.apache.spark._
import org.apache.spark.rdd._
import org.apache.spark.SparkContext._
import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}

val input = sc.textFile("/Users/rkita/Documents/Learning/random/spark/MLlib/examples/text8",4).map(line => line.split(" ").toSeq)

val word2vec = new Word2Vec()

val model = word2vec.fit(input)

val synonyms = model.findSynonyms("china", 40)

for((synonym, cosineSimilarity) <- synonyms) {
  println(s"$synonym $cosineSimilarity")
}

// Save and load model
model.save(sc, "myModelPath")
val sameModel = Word2VecModel.load(sc, "myModelPath")

我正在我的 mac(2 核，8GB RAM)上使用 Spark，我想我已经在我的 spark-env.sh 文件中正确设置了内存分配，如下所示:

export SPARK_EXECUTOR_MEMORY=4g
export SPARK_WORKER_MEMORY=4g

当我尝试拟合模型时，我不断收到 Java 堆错误。我在 python 中也得到了相同的结果。我也使用 JAVA_OPTS 增加了 java 内存大小。

该文件只有 100MB，所以我认为我的内存设置不正确，但我不确定这是否是根本原因。

有没有其他人在笔记本电脑上尝试过这个例子？

我不能把文件放在我们公司的服务器上，因为我们不应该导入外部数据，所以我只能在我的个人笔记本电脑上工作。如果您有任何建议，我将不胜感激。谢谢!

Answer 1

首先，我是Spark的新手，所以其他人可能有更快或更好的解决方案。
我在运行此示例代码时遇到了同样的困难。
我设法使它工作，主要是通过:

在我的机器上运行我自己的 Spark 集群:使用 Spark 安装的/sbin/目录中的启动脚本。为此，您必须根据需要配置 conf/spark-env.sh 文件。不要将 127.0.0.1 IP 用于 Spark。

将Scala代码编译打包为jar(sbt包)，然后提供给集群(参见Scala代码中的addJar(...))。似乎可以使用类路径/额外类路径向 Spark 提供编译代码，但我还没有尝试过。

设置执行程序内存和驱动程序内存(参见 Scala 代码)

spark-env.sh:

export SPARK_MASTER_IP=192.168.1.53
export SPARK_MASTER_PORT=7077
export SPARK_MASTER_WEBUI_PORT=8080

export SPARK_DAEMON_MEMORY=1G
# Worker : 1 by server
# Number of worker instances to run on each machine (default: 1). 
# You can make this more than 1 if you have have very large machines and would like multiple Spark worker processes. 
# If you do set this, make sure to also set SPARK_WORKER_CORES explicitly to limit the cores per worker, 
# or else each worker will try to use all the cores.
export SPARK_WORKER_INSTANCES=2
# Total number of cores to allow Spark applications to use on the machine (default: all available cores).
export SPARK_WORKER_CORES=7

#Total amount of memory to allow Spark applications to use on the machine, e.g. 1000m, 2g 
# (default: total memory minus 1 GB); 
# note that each application's individual memory is configured using its spark.executor.memory property.
export SPARK_WORKER_MEMORY=8G
export SPARK_WORKER_DIR=/tmp

# Executor : 1 by application run on the server
# export SPARK_EXECUTOR_INSTANCES=4
# export SPARK_EXECUTOR_MEMORY=4G

export SPARK_SCALA_VERSION="2.10"

Scala 文件来运行示例:

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}

object SparkDemo {

  def log[A](key:String)(job : =>A) = {
    val start = System.currentTimeMillis
    val output = job
    println("===> %s in %s seconds"
      .format(key, (System.currentTimeMillis - start) / 1000.0))
    output
  }

  def main(args: Array[String]):Unit ={

    val modelName ="w2vModel"

    val sc = new SparkContext(
      new SparkConf()
      .setAppName("SparkDemo")
      .set("spark.executor.memory", "8G")
      .set("spark.driver.maxResultSize", "16G")
      .setMaster("spark://192.168.1.53:7077") // ip of the spark master.
      // .setMaster("local[2]") // does not work... workers loose contact with the master after 120s
    )

    // take a look into target folder if you are unsure how the jar is named
    // onliner to compile / run : sbt package && sbt run
    sc.addJar("./target/scala-2.10/sparkling_2.10-0.1.jar")

    val input = sc.textFile("./text8").map(line => line.split(" ").toSeq)

    val word2vec = new Word2Vec()

    val model = log("compute model") { word2vec.fit(input) }
    log ("save model") { model.save(sc, modelName) }

    val synonyms = model.findSynonyms("china", 40)
    for((synonym, cosineSimilarity) <- synonyms) {
      println(s"$synonym $cosineSimilarity")
    }

    val model2 = log("reload model") { Word2VecModel.load(sc, modelName) }
  }
}