apache-spark - Spark : Poor performance on distributed system. 如何改进>

Question

我写了一个简单的Spark程序，想把它部署到分布式服务器上。这很简单:

获取数据->整理数据->训练数据->重新申请看训练结果。

输入数据只有 10K 行，具有 3 个特征。我首先在我的本地机器上运行，使用“local[*]”。它只运行大约 3 分钟。现在当我部署到一个集群时，它运行得非常慢:半小时没有完成。在训练阶段变得很慢。

我很好奇，如果我做错了什么。请帮我检查一下。我使用 Spark 1.6.1。

我提交:

spark-submit --packages com.databricks:spark-csv_2.11:1.5.0  orderprediction_2.11-1.0.jar --driver-cores 1 --driver-memory 4g --executor-cores 8 --executor-memory 4g

代码在这里:

 def main(args: Array[String]) {
    // Set the log level to only print errors
    Logger.getLogger("org").setLevel(Level.ERROR)

    val conf = new SparkConf()
        .setAppName("My Prediction")
        //.setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val data = sqlContext.read
        .option("header","true")
        .option("delimiter", "\t")
        .format("com.databricks.spark.csv")
        .option("inferSchema","true")
        .load("mydata.txt")

    data.printSchema()
    data.show()

    val dataDF = data.toDF().filter("clicks >=10")
    dataDF.show()

    val assembler = new VectorAssembler()
      .setInputCols(Array("feature1", "feature2", "feature3"))
      .setOutputCol("features")

    val trainset = assembler.transform(dataDF).select("target", "features")
    trainset.printSchema()
    val trainset2 = trainset.withColumnRenamed("target", "label")

    trainset2.printSchema()
    val trainset3 = trainset2.withColumn("label", trainset2.col("label").cast(DataTypes.DoubleType))
    trainset3.cache() // cache data into memory
    trainset3.printSchema()
    trainset3.show()

    // Train a RandomForest model.

    println("training Random Forest")

    val rf = new RandomForestRegressor()
      .setLabelCol("label")
      .setFeaturesCol("features")
      .setNumTrees(1000)

    val rfmodel = rf.fit(trainset3)

    println("prediction")
    val result = rfmodel.transform(trainset3)

    result.show()
}

更新:经过排查，发现卡在了

collectAsMap at RandomForest.scala:525

这条线已经用了1.1个小时了，还未完成。数据，我相信只有几兆字节。

Answer 1

您正在构建一个由 1000 棵随机树组成的随机森林，它将训练 1000 个实例。

在代码中，collectAsMap 是第一个操作，而其余所有操作都是转换(惰性求值)。所以当您看到它卡在那一行时，那是因为现在所有的 map 、平面 map 、过滤器、groupBy、等都被评估了。