如何执行 Spark 数据加载并有效地执行转换答案

【问题标题】：How to perfom spark data load and perform a transformation efficently如何执行 Spark 数据加载并有效地执行转换
【发布时间】：2022-07-07 13:20:17
【问题描述】：

我一直在尝试使用 apache spark 运行应用程序，通常它从一系列数据库表中获取信息并创建一个包含带有集合的列的数据集。这进入使用 lambda 的 map()。目前它从数据库中提取了 140,000 条复合记录。

为了使数据加载数据集，我们通过五个数据集连接包含每个表的信息，其中大部分是与第一个数据集的键相关的列表

在完成简单的计算之后，取值并对其进行操作，没有嵌套循环或不必要的循环

正在使用的集群有 9 个工作器，每个工作器有 8 个内核和 64 GB 内存。 spark 配置是默认配置，尽管如此，日志始终显示相同的异常：

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 0

配置是下一个

config.set("spark.driver.maxResultSize", "0");
config.set("spark.speculation", "true");
config.set("spark.sql.broadcastTimeout", "600");
config.set("spark.driver.memory","64g");
config.set("spark.executor.memory","64g");

【问题讨论】：

你能在这里添加你的代码

标签： java apache apache-spark pyspark apache-spark-sql

【解决方案1】：

要么添加交换，要么配置工作程序/执行程序以使用更少的内存，并使用 MEMORY_AND_DISK 存储级别进行多个持久化。

解决这个问题的方法是执行以下步骤：

增加随机分区的数量：

--conf spark.sql.shuffle.partitions=
如果您不持久化或缓存任何数据帧，您可以在增加 shuffle 分区数量后减少 spark 内存的存储部分。默认情况下，存储部分是 0.5 ，执行部分也是 0.5 。要减少存储部分，您可以在 spark-submit 命令中设置以下配置

--conf spark.memory.storageFraction=0.3
除了以上两点，你还可以设置执行器开销内存。

--conf spark.executor.memoryOverhead=2g

这是用于虚拟机开销的堆外内存，interned 字符串等
除此之外，您可以通过将 maxFilesPerTrigger 设置为较小的值（例如 10）来限制特定微批处理中处理的文件数。

【讨论】：