【问题标题】:Spark jobs finishes but application takes time to closeSpark 作业完成,但应用程序需要时间才能关闭
【发布时间】:2016-05-02 07:52:26
【问题描述】:

使用 scala 运行 spark 作业,正如预期的那样,所有作业都按时完成,但不知何故,一些 INFO 日志会在作业停止前打印 20-25 分钟。

张贴一些 UI 截图,可以帮助理解问题。

  1. 以下是 4 个阶段所花费的时间:

  1. 以下是连续作业 ID 之间的时间

我不明白为什么在两个工作 ID 之间花费了这么多时间。

以下是我的代码 sn-p:

    val sc = new SparkContext(conf)
for (x <- 0 to 10) {
  val zz = getFilesList(lin);
  val links = zz._1
  val path = zz._2
  lin = zz._3
  val z = sc.textFile(links.mkString(",")).map(t => t.split('\t')).filter(t => t(4) == "xx" && t(6) == "x").map(t => titan2(t)).filter(t => t.length > 35).map(t => ((t(34)), (t(35), t(5), t(32), t(33))))
  val way_nodes = sc.textFile(way_source).map(t => t.split(";")).map(t => (t(0), t(1)));
  val t = z.join(way_nodes).map(t => (t._2._1._2, Array(Array(t._2._1._2, t._2._1._3, t._2._1._4, t._2._1._1, t._2._2)))).reduceByKey((t, y) => t ++ y).map(t => process(t)).flatMap(t => t).combineByKey(createTimeCombiner, timeCombiner, timeMerger).map(averagingFunction).map(t => t._1 + "," + t._2)
  t.saveAsTextFile(path)
}
sc.stop()

更多跟进:spark-1.4.1 saveAsTextFile to S3 is very slow on emr-4.0.0

【问题讨论】:

  • 我通常建议使用 Databricks 的 spark-csv 包而不是 saveAsTextFile,但除此之外,您运行的是哪个版本的 Spark?
  • saveAsTextFile 的优点是我可以直接将所有内容保存在 s3 上,不确定 spark-csv 包 databricks 是如何工作的。感谢您的一些指导,无论如何都会调查它。火花 - 1.4.1 斯卡拉 - 2.10.6

标签: scala amazon-s3 apache-spark


【解决方案1】:

正如我发表的评论,我建议使用 spark-csv 包而不是 sc.saveAsTextFile,使用该包直接写入 s3 没有问题 :)

我不知道你是用 s3 还是 s3n,但也许可以尝试切换。我在 Spark 1.5.2 (EMR-4.2) 上使用 s3a 时遇到了问题,其中写入一直超时,切换回 s3 解决了这个问题,所以值得一试。

应该加快写入 s3 的其他几件事是使用 DirectOutputCommiter

conf.set("spark.hadoop.mapred.output.committer.class","com.appsflyer.spark.DirectOutputCommitter")

并禁用 _SUCCESS 文件的生成:

sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

请注意,必须在 SparkContext 的 hadoop 配置上而不是在 SparkConf 上设置禁用 _SUCCESS 文件。

我希望这会有所帮助。

【讨论】:

【解决方案2】:

我最终升级了我的 spark 版本并解决了问题。

【讨论】:

【解决方案3】:

我在将文件写入 S3 时遇到了同样的问题。我用的是spark 2.0版本,只是为了给你验证答案的更新代码

在 Spark 2.0 中你可以使用,

val spark = SparkSession.builder().master("local[*]").appName("App_name").getOrCreate()

spark.conf.set("spark.hadoop.mapred.output.committer.class","com.appsflyer.spark.DirectOutputCommitter")
spark.conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

这解决了我的工作被击中的问题

【讨论】:

猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2017-12-31
  • 1970-01-01
  • 2017-05-16
  • 2022-12-23
  • 2018-02-07
相关资源
最近更新 更多