Spark故障:由于:org.apache.spark.shuffle.FetchFailedException:框架过大:5454002341

6yt4nkrj  于 2023-06-06  发布在  Apache
关注(0)|答案(6)|浏览(371)

我正在为一个确定父子关系的表生成层次结构。
以下是使用的配置,即使在获得关于过大帧的错误之后:

Spark属性

--conf spark.yarn.executor.memoryOverhead=1024mb \
--conf yarn.nodemanager.resource.memory-mb=12288mb \
--driver-memory 32g \
--driver-cores  8 \
--executor-cores 32 \
--num-executors 8 \
--executor-memory 256g \
--conf spark.maxRemoteBlockSizeFetchToMem=15g
import org.apache.log4j.{Level, Logger};
import org.apache.spark.SparkContext;
import org.apache.spark.sql.{DataFrame, SparkSession};
import org.apache.spark.sql.functions._;
import org.apache.spark.sql.expressions._;

lazy val sparkSession = SparkSession.builder.enableHiveSupport().getOrCreate();

import spark.implicits._;

val hiveEmp: DataFrame = sparkSession.sql("select * from db.employee");
hiveEmp.repartition(300);
import org.apache.spark.sql.functions._;

val nestedLevel = 3;

val empHierarchy = (1 to nestedLevel).foldLeft(hiveEmp.as("wd0")) { (wDf, i) =>
  val j = i - 1
  wDf.join(hiveEmp.as(s"wd$i"), col(s"wd$j.parent_id".trim) === col(s"wd$i.id".trim), "left_outer")
}.select(
  col("wd0.id") :: col("wd0.parent_id") ::
  col("wd0.amount").as("amount") :: col("wd0.payment_id").as("payment_id") :: (
    (1 to nestedLevel).toList.map(i => col(s"wd$i.amount").as(s"amount_$i")) :::
    (1 to nestedLevel).toList.map(i => col(s"wd$i.payment_id").as(s"payment_id_$i"))

  ): _*);

empHierarchy.write.saveAsTable("employee4");

错误

Caused by: org.apache.spark.SparkException: Task failed while writing rows
   at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:204)
   at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$3.apply(FileFormatWriter.scala:129)
   at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$3.apply(FileFormatWriter.scala:128)
   at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
   at org.apache.spark.scheduler.Task.run(Task.scala:99)
   at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
   ... 3 more
Caused by: org.apache.spark.shuffle.FetchFailedException: Too large frame: 5454002341
   at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:361)
   at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:336)
j5fpnvbx

j5fpnvbx1#

使用此spark配置,spark.maxRemoteBlockSizeFetchToMem < 2g
由于> 2G分区有很多问题(不能shuffle,不能在磁盘上缓存),因此它抛出failedfetchedexception太大的 Dataframe 。

xmjla07d

xmjla07d2#

苏雷什说得对以下是他的回答的一个更好的文档化和格式化版本,其中包含一些有用的背景信息:

如果您使用的是2.2.x或2.3.x版本,则可以通过将config的值设置为Int.MaxValue - 512来实现相同的效果,即通过设置spark.maxRemoteBlockSizeFetchToMem=2147483135。请参阅此处,了解截至2019年9月使用的默认值。

bvuwiixz

bvuwiixz3#

这意味着数据集分区的大小是巨大的。您需要将数据集重新分区到更多分区。
你可以使用

df.repartition(n)

这里,n取决于数据集的大小。

ruarlubt

ruarlubt4#

在尝试回填几年的数据时遇到了完全相同的错误。事实证明,这是因为您的分区大小> 2GB。
1.您可以增加分区的数量(使用repartition()),使您的分区小于2GB。(保持您的分区接近128 mb到256 mb,即接近HDFS数据块大小)
1.或者你也可以像上面提到的那样将shuffle限制提高到> 2GB。(避免)。此外,数据量大的分区会导致任务需要很长时间才能完成。
注意:repartition(n)将导致在写入s3/hdfs时每个分区有n个部分文件。
请阅读以下内容以获取更多信息:http://www.russellspitzer.com/2018/05/10/SparkPartitions/

pxy2qtax

pxy2qtax5#

我在处理~ 700 GB数据集时遇到了同样的问题。减少spark.maxRemoteBlockSizeFetchToMem在我的情况下没有帮助。此外,我无法增加分区的数量。
做以下工作对我来说很有用:
1.增加spark.network.timeout(Spark 2.3中的默认值为120秒),这会影响以下内容:

spark.core.connection.ack.wait.timeout
spark.storage.blockManagerSlaveTimeoutMs
spark.shuffle.io.connectionTimeout
spark.rpc.askTimeout
spark.rpc.lookupTimeout

1.设置spark.network.timeout=600s(Spark 2.3中默认为120 s)
1.设置spark.io.compression.lz4.blockSize=512k(Spark 2.3中默认为32 k)
1.设置spark.shuffle.file.buffer=1024k(Spark 2.3中默认为32 k)

vcirk6k6

vcirk6k66#

下面的配置为我工作。
1.保持spark.sql.shuffle.partitions和spark.default.parallelism相同数量

  1. spark.maxRemoteBlockSizeFetchToMem <2GB
  2. spark.shuffle.spill.compress and spark.shuffle.compress to“true..
    “spark.maxRemoteBlockSizeFetchToMem”:“2147483135”,“spark.sql.shuffle.partitions”:“3000”,“spark.default.parallelism”:“3000”,“spark.shuffle.spill.compress”:“true”,“spark.shuffle.compress”:“真”

相关问题