3

我们正在尝试实现一个简单的 spark 作业，它读取 CSV 文件（1 行数据）并使用预构建的随机森林模型对象进行预测。这项工作不包括任何数据预处理或数据操作。

我们以独立模式运行 spark，应用程序在本地运行。配置如下： RAM：8GB 内存：40GB 内核数：2 Spark 版本：1.5.2 Scala 版本：2.10.5 输入文件大小：1KB（1行数据） 模型文件大小：1,595 KB（400棵树）随机森林）

目前，spark-submit 中的实现大约需要 13 秒。然而，运行时间是这个应用程序的一个巨大问题，因此

  1. 有没有办法优化代码以将运行时间缩短到 1 或 2 秒？（高优先级）

  2. 我们注意到实际代码的执行大约需要 7-8 秒，而启动和设置上下文大约需要 5-6 秒，所以有没有办法在我们运行 spark-submit 时保持 spark 上下文运行。

这是应用程序代码

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

object RF_model_App {
  def main(args: Array[String]) {

val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature4.{RandomForestfeature4Model, RandomForestClassifier}
import org.apache.spark.ml.evaluation.Multiclassfeature4Evaluator
import org.apache.spark.ml.feature.{IndexToString, StringIndexer, VectorIndexer}
import org.apache.spark.sql.functions.udf
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.feature.StringIndexer
import sqlContext.implicits._
val Test = sqlContext.read.format("com.databricks.spark.csv").option("header","true").load("/home/ubuntu/Test.csv")
Test.registerTempTable("Test")
val model_L1 = sc.objectFile[RandomForestfeature4Model]("/home/ubuntu/RF_L1.model").first()

val toInt = udf[Int, String]( _.toInt)
val toDouble = udf[Double, String]( _.toDouble)
val featureDf = Test.withColumn("id1", toInt(Test("id1")))  .withColumn("id2", toInt(Test("id2")))  .withColumn("id3", toInt(Test("id3")))  .withColumn("id4", toInt(Test("id4")))   .withColumn("feature3", toInt(Test("feature3")))   .withColumn("feature9", toInt(Test("feature9")))    .withColumn("feature10", toInt(Test("feature10")))  .withColumn("feature12", toInt(Test("feature12")))  .withColumn("feature14", toDouble(Test("feature14")))   .withColumn("feature15", toDouble(Test("feature15")))   .withColumn("feature16", toInt(Test("feature16")))  .withColumn("feature17", toDouble(Test("feature17")))   .withColumn("feature18", toInt(Test("feature18")))

val feature4_index = new StringIndexer()  .setInputCol("feature4")  .setOutputCol("feature4_index")
val feature6_index = new StringIndexer()  .setInputCol("feature6")  .setOutputCol("feature6_index")
val feature11_index = new StringIndexer()  .setInputCol("feature11")  .setOutputCol("feature11_index")
val feature8_index = new StringIndexer()  .setInputCol("feature8")  .setOutputCol("feature8_index")
val feature13_index = new StringIndexer()  .setInputCol("feature13")  .setOutputCol("feature13_index")
val feature2_index = new StringIndexer()  .setInputCol("feature2")  .setOutputCol("feature2_index")
val feature5_index = new StringIndexer()  .setInputCol("feature5")  .setOutputCol("feature5_index")
val feature7_index = new StringIndexer()  .setInputCol("feature7")  .setOutputCol("feature7_index")
val vectorizer_L1 =  new VectorAssembler()  .setInputCols(Array("feature3",  "feature2_index", "feature6_index", "feature4_index", "feature8_index", "feature7_index", "feature5_index", "feature10", "feature9", "feature12", "feature11_index", "feature13_index", "feature14", "feature15", "feature18", "feature17", "feature16")).setOutputCol("features_L1")
val feature_pipeline_L1 = new Pipeline()  .setStages(Array( feature4_index, feature6_index, feature11_index,feature8_index, feature13_index,  feature2_index, feature5_index, feature7_index,vectorizer_L1))
val testPredict= feature_pipeline_L1.fit(featureDf).transform(featureDf)
val getPOne = udf((v: org.apache.spark.mllib.linalg.Vector) => v(1))
val getid2 = udf((v: Int) => v)
val L1_output = model_L1.transform(testPredict).select(getid2($"id2") as "id2",getid2($"prediction") as "L1_prediction",getPOne($"probability") as "probability")

L1_output.repartition(1).write.format("com.databricks.spark.csv").option("header", "true").mode("overwrite").save("/home/L1_output")

  }
};
4

1 回答 1

1

让我们从完全错误的事情开始：

  • 您使用的功能机制不正确。StringIndexer根据数据的分布分配索引，因此同一记录将根据其他记录具有不同的编码。您应该使用相同的StringIndexerModel(-s) 进行训练、测试和预测。
  • val getid2 = udf((v: Int) => v)只是一个昂贵的身份。

执着的SparkContext

有多种工具可以保持持久的上下文，包括job-serveror Livy。

最后，您可以简单地使用 Spark Streaming 并直接处理数据。

洗牌

您还repartition用于创建单个文件，因此我想是一个文件 CSV。这个动作非常昂贵，但在定义上，它会随机重新排列 RDD 中的数据以创建更多或更少的分区并在它们之间进行平衡。这总是对网络上的所有数据进行洗牌。

其他注意事项：

如果延迟很重要，并且您只使用一台低性能机器，则根本不要使用 Spark。在这里没有任何收获。在这种情况下，一个好的本地图书馆可以做得更好。

备注：

我们不会访问您的数据或硬件，因此任何要求，例如将时间缩短到 7 秒都是完全没有意义的。

于 2016-02-26T15:00:26.627 回答