由于 space 问题导致 Spark 作业失败
Spark job failing due to space issue
我正在使用 pyspark 在 Spark 中编写批处理程序。
以下是输入文件及其大小
base-track.dat (3.9g)
base-attribute-link.dat (18g)
base-release.dat (543m)
这些是每行一条记录的文本文件，每个字段由一个特殊字符分隔（参考代码）
我正在对属性 link 执行一些过滤操作并将它们分组并与其他表连接。
我正在通过 spark-submit 将此程序提交到一个由 Ambari 管理的具有 9 个数据节点的 Hadoop 集群。
每个数据节点包含 140 GB RAM 和 3.5 TB 磁盘 space。
以下是我的pyspark代码
import sys
from pyspark import SparkContext
from pyspark.sql import SQLContext, Row
if __name__ == "__main__":
sc = SparkContext(appName = "Tracks")
sqlContext = SQLContext(sc)
#Load base-track
track = sc.textFile("base-track/input").map(lambda row: row.split(u'\u0001'))
#Load base-attribute-link
attlnk = sc.textFile("base-attribute-link/input").map(lambda row: row.split(u'\u0001'))
#Load base-release
release = sc.textFile("base-release/input").map(lambda row: row.split(u'\u0001'))
attlnk = attlnk.filter(lambda row: row[2] == 'MA0000000162')
attlnkg = attlnk.groupBy(lambda row: row[1])
attlnkmax = attlnkg.map( lambda t: (t[0],max([v[4] for v in t[1]])) )
alg = attlnkmax.map(lambda r: Row(al_objectid=r[0],al_value=r[1]))
aldf = alg.toDF()
track = track.map(lambda r:Row(t_tag = r[0], t_trackid= r[1], t_releaseid= r[2], t_songid = r[3], t_med= r[4], t_ph = r[5], t_tn = r[5], t_title= r[5], t_part= r[6], t_dur = r[7], t_pick = r[8], t_amgclid = r[9], t_amgpopid = r[10], t_compid = r[11], t_muzid = r[12], t_perfid= r[13], t_albumid = r[14]))
trackdf = track.toDF()
release = release.map(lambda r:Row(r_tag = r[0], r_relid = r[1], r_albumid = r[2], r_mediafmtid = r[3], r_prodfmtid = r[4], r_reldate = r[5], r_prodcode = r[6], r_prodtypeid = r[7], r_label = r[8], r_relyear = r[9], r_ispurch = r[10], r_amgclassid = r[11], r_amgpopid = r[12], r_eanid = r[13], r_upcid = r[14]))
releasedf = release.toDF()
trackaldf = trackdf.join(aldf, trackdf['t_trackid'] == aldf['al_objectid'], 'left_outer')
tracksdf = trackaldf.join(releasedf, trackaldf['t_releaseid'] == releasedf['r_relid'])
tracksdf = tracksdf.select('t_trackid', 't_releaseid', 't_songid', 't_med', 't_ph', 't_tn', 't_title', 't_part', 't_dur', 't_pick', 't_amgclid', 't_amgpopid', 't_compid', 't_muzid', 'al_objectid', 't_perfid', 't_albumid', 'r_label')
tracksdf.rdd.map(lambda x: u"\u0001".join(map(str, x))).coalesce(100).saveAsTextFile("tracks-out")
在尝试执行此操作时出现以下错误。
ERROR DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file /tmp/blockmgr-d88c631e-cec3-4b83-8af6-a38b109b5e3b/0e/temp_shuffle_7dbda3ac-48b1-4c4a-89c7-64eb5d858d90
java.io.IOException: No space left on device
at java.io.FileOutputStream.writeBytes(Native Method)
at java.io.FileOutputStream.write(FileOutputStream.java:326)
at org.apache.spark.storage.TimeTrackingOutputStream.write(TimeTrackingOutputStream.java:58)
at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
at org.xerial.snappy.SnappyOutputStream.flush(SnappyOutputStream.java:336)
at org.apache.spark.io.SnappyOutputStreamWrapper.flush(CompressionCodec.scala:209)
at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:141)
at java.io.DataOutputStream.flush(DataOutputStream.java:123)
at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon.flush(UnsafeRowSerializer.scala:83)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply$mcV$sp(DiskBlockObjectWriter.scala:157)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply(DiskBlockObjectWriter.scala:154)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply(DiskBlockObjectWriter.scala:154)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1239)
at org.apache.spark.storage.DiskBlockObjectWriter.revertPartialWritesAndClose(DiskBlockObjectWriter.scala:161)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.stop(BypassMergeSortShuffleWriter.java:232)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
SO 上有几个问题， and here 与同一问题相关。
以上两个问题我试过的就是这个。
我试图将 spark.yarn.executor.memoryOverhead 从 384 MB 增加到 4GB。
SPARK_JAVA_OPTS+=" -Dspark.local.dir=/mnt/spark,/mnt2/spark -Dhadoop.tmp.dir=/mnt/ephemeral-hdfs"
export SPARK_JAVA_OPTS
第一个没有任何效果。如果我添加 java opts.
，我会收到错误消息 /mnt 目录不存在
在多个论坛（包括 databricks）上阅读了有关此问题的内容后，我有一些模糊的想法，认为该作业正在尝试创建临时文件，作为每个集群节点 /tmp 上随机播放的一部分，并耗尽 space。在每个集群节点上，我们为 tmp 目录所在的根 (/) 分区分配了 100 GB。
我已经努力了一个多月，通过使用各种 spark 配置参数来执行此操作。作为调整的一部分，我将 spark.driver 和 spark.executor 内存增加到 16g，后来又增加到 64g。还将 spark yarn 执行器内存增加到 4GB。不幸的是 none 这可以解决 space 问题。
任何关于如何进一步进行的指导都会有很大帮助。
[Edit-1] 我刚刚检查了所有机器上根目录的磁盘 space，我们集群中的 9 个节点中有 7 个有 100+GB分配给根目录，但在 2 个节点上只分配了 10 GB，它们只剩下 6+GB。这可能会导致磁盘 space 问题，如果可以扩展根目录的大小，我将不得不与我们的 IT 团队核实。
[Edit-2] 我与 IT 团队合作将所有机器上的根分区大小扩展到 100+GB，但问题仍然存在，可能是 100GB /tmp space 也不足以完成这项工作。我估计这个作业的输出大约是 4.6GB。
考虑到您的错误的性质以及您正在对数十 GB 的数据执行大型连接，其中 spark worker 会在 shuffle 时将中间数据写入磁盘，100GB 的磁盘似乎不够用.我建议为默认 worker_dir 和 local_dirs 分配更多磁盘，方法是将它们安装到更大的磁盘或配置更大的根磁盘。另外，请注意，如果 spark 未正确关闭，则此中间数据可能会持续存在并占用工作节点上的大量 space。因此，您可能必须检查这些目录并删除所有陈旧文件。如果您在 AWS r3、c3 或具有大型临时 SSD 磁盘的类似实例类型上 运行 spark-standalone，我建议 mounting those disks to say "mnt" and "mnt2" and configuring spark scratch space 指向这些装载，而不是（通常）较小的根体积。例如：
SPARK_LOCAL_DIRS=/mnt
SPARK_WORKER_DIR=/mnt2
我发现我不是在向集群提交 spark 作业，而是在一台机器上提交，因此磁盘 space 出现问题。我总是通过以下方式提交脚本
spark-submit tracks.py
因为我希望我的脚本在Hadoop集群上执行并且使用Yarn作为资源管理器，所以我将我的提交命令更改为以下，然后它工作正常。
spark-submit --master yarn --deploy-mode cluster tracks.py
我正在使用 pyspark 在 Spark 中编写批处理程序。 以下是输入文件及其大小
base-track.dat (3.9g)
base-attribute-link.dat (18g)
base-release.dat (543m)
这些是每行一条记录的文本文件，每个字段由一个特殊字符分隔（参考代码）
我正在对属性 link 执行一些过滤操作并将它们分组并与其他表连接。
我正在通过 spark-submit 将此程序提交到一个由 Ambari 管理的具有 9 个数据节点的 Hadoop 集群。 每个数据节点包含 140 GB RAM 和 3.5 TB 磁盘 space。
以下是我的pyspark代码
import sys
from pyspark import SparkContext
from pyspark.sql import SQLContext, Row
if __name__ == "__main__":
sc = SparkContext(appName = "Tracks")
sqlContext = SQLContext(sc)
#Load base-track
track = sc.textFile("base-track/input").map(lambda row: row.split(u'\u0001'))
#Load base-attribute-link
attlnk = sc.textFile("base-attribute-link/input").map(lambda row: row.split(u'\u0001'))
#Load base-release
release = sc.textFile("base-release/input").map(lambda row: row.split(u'\u0001'))
attlnk = attlnk.filter(lambda row: row[2] == 'MA0000000162')
attlnkg = attlnk.groupBy(lambda row: row[1])
attlnkmax = attlnkg.map( lambda t: (t[0],max([v[4] for v in t[1]])) )
alg = attlnkmax.map(lambda r: Row(al_objectid=r[0],al_value=r[1]))
aldf = alg.toDF()
track = track.map(lambda r:Row(t_tag = r[0], t_trackid= r[1], t_releaseid= r[2], t_songid = r[3], t_med= r[4], t_ph = r[5], t_tn = r[5], t_title= r[5], t_part= r[6], t_dur = r[7], t_pick = r[8], t_amgclid = r[9], t_amgpopid = r[10], t_compid = r[11], t_muzid = r[12], t_perfid= r[13], t_albumid = r[14]))
trackdf = track.toDF()
release = release.map(lambda r:Row(r_tag = r[0], r_relid = r[1], r_albumid = r[2], r_mediafmtid = r[3], r_prodfmtid = r[4], r_reldate = r[5], r_prodcode = r[6], r_prodtypeid = r[7], r_label = r[8], r_relyear = r[9], r_ispurch = r[10], r_amgclassid = r[11], r_amgpopid = r[12], r_eanid = r[13], r_upcid = r[14]))
releasedf = release.toDF()
trackaldf = trackdf.join(aldf, trackdf['t_trackid'] == aldf['al_objectid'], 'left_outer')
tracksdf = trackaldf.join(releasedf, trackaldf['t_releaseid'] == releasedf['r_relid'])
tracksdf = tracksdf.select('t_trackid', 't_releaseid', 't_songid', 't_med', 't_ph', 't_tn', 't_title', 't_part', 't_dur', 't_pick', 't_amgclid', 't_amgpopid', 't_compid', 't_muzid', 'al_objectid', 't_perfid', 't_albumid', 'r_label')
tracksdf.rdd.map(lambda x: u"\u0001".join(map(str, x))).coalesce(100).saveAsTextFile("tracks-out")
在尝试执行此操作时出现以下错误。
ERROR DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file /tmp/blockmgr-d88c631e-cec3-4b83-8af6-a38b109b5e3b/0e/temp_shuffle_7dbda3ac-48b1-4c4a-89c7-64eb5d858d90
java.io.IOException: No space left on device
at java.io.FileOutputStream.writeBytes(Native Method)
at java.io.FileOutputStream.write(FileOutputStream.java:326)
at org.apache.spark.storage.TimeTrackingOutputStream.write(TimeTrackingOutputStream.java:58)
at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
at org.xerial.snappy.SnappyOutputStream.flush(SnappyOutputStream.java:336)
at org.apache.spark.io.SnappyOutputStreamWrapper.flush(CompressionCodec.scala:209)
at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:141)
at java.io.DataOutputStream.flush(DataOutputStream.java:123)
at org.apache.spark.sql.execution.UnsafeRowSerializerInstance$$anon.flush(UnsafeRowSerializer.scala:83)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply$mcV$sp(DiskBlockObjectWriter.scala:157)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply(DiskBlockObjectWriter.scala:154)
at org.apache.spark.storage.DiskBlockObjectWriter$$anonfun$revertPartialWritesAndClose.apply(DiskBlockObjectWriter.scala:154)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1239)
at org.apache.spark.storage.DiskBlockObjectWriter.revertPartialWritesAndClose(DiskBlockObjectWriter.scala:161)
at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.stop(BypassMergeSortShuffleWriter.java:232)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:79)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
SO 上有几个问题，
以上两个问题我试过的就是这个。 我试图将 spark.yarn.executor.memoryOverhead 从 384 MB 增加到 4GB。
SPARK_JAVA_OPTS+=" -Dspark.local.dir=/mnt/spark,/mnt2/spark -Dhadoop.tmp.dir=/mnt/ephemeral-hdfs"
export SPARK_JAVA_OPTS
第一个没有任何效果。如果我添加 java opts.
，我会收到错误消息 /mnt 目录不存在在多个论坛（包括 databricks）上阅读了有关此问题的内容后，我有一些模糊的想法，认为该作业正在尝试创建临时文件，作为每个集群节点 /tmp 上随机播放的一部分，并耗尽 space。在每个集群节点上，我们为 tmp 目录所在的根 (/) 分区分配了 100 GB。
我已经努力了一个多月，通过使用各种 spark 配置参数来执行此操作。作为调整的一部分，我将 spark.driver 和 spark.executor 内存增加到 16g，后来又增加到 64g。还将 spark yarn 执行器内存增加到 4GB。不幸的是 none 这可以解决 space 问题。
任何关于如何进一步进行的指导都会有很大帮助。
[Edit-1] 我刚刚检查了所有机器上根目录的磁盘 space，我们集群中的 9 个节点中有 7 个有 100+GB分配给根目录，但在 2 个节点上只分配了 10 GB，它们只剩下 6+GB。这可能会导致磁盘 space 问题，如果可以扩展根目录的大小，我将不得不与我们的 IT 团队核实。
[Edit-2] 我与 IT 团队合作将所有机器上的根分区大小扩展到 100+GB，但问题仍然存在，可能是 100GB /tmp space 也不足以完成这项工作。我估计这个作业的输出大约是 4.6GB。
考虑到您的错误的性质以及您正在对数十 GB 的数据执行大型连接，其中 spark worker 会在 shuffle 时将中间数据写入磁盘，100GB 的磁盘似乎不够用.我建议为默认 worker_dir 和 local_dirs 分配更多磁盘，方法是将它们安装到更大的磁盘或配置更大的根磁盘。另外，请注意，如果 spark 未正确关闭，则此中间数据可能会持续存在并占用工作节点上的大量 space。因此，您可能必须检查这些目录并删除所有陈旧文件。如果您在 AWS r3、c3 或具有大型临时 SSD 磁盘的类似实例类型上 运行 spark-standalone，我建议 mounting those disks to say "mnt" and "mnt2" and configuring spark scratch space 指向这些装载，而不是（通常）较小的根体积。例如：
SPARK_LOCAL_DIRS=/mnt
SPARK_WORKER_DIR=/mnt2
我发现我不是在向集群提交 spark 作业，而是在一台机器上提交，因此磁盘 space 出现问题。我总是通过以下方式提交脚本
spark-submit tracks.py
因为我希望我的脚本在Hadoop集群上执行并且使用Yarn作为资源管理器，所以我将我的提交命令更改为以下，然后它工作正常。
spark-submit --master yarn --deploy-mode cluster tracks.py