【发布时间】：2020-07-26 14:02:38
【问题描述】：
使用 Pyspark，尝试将大量 json 文件从 S3 加载到数据帧时出现错误。该错误似乎取决于我的 spark 会话使用的驱动程序内存。错误消息显示为“java.lang.OutOfMemoryError：GC 开销限制超出”。我花了很多时间进行在线研究，但我找不到任何可以指出这个错误的确切原因的东西。请在下面找到完整的错误消息和代码。
如果能提供任何帮助，我将不胜感激！
驱动环境
我在 jupyterlab 中运行的 jupyter notebook 中使用 pyspark，它本身在具有 30GB 可用内存的 EC2 实例上运行。
Sparksession 资源
spark.executor.memory：3GB
spark.executor.cores：2
spark.driver.memory：5GB
spark.cores.max：300
数据
我正在尝试读取存储在 S3 中的大约 500k json 文件，总数据大小为 100+GB。每个文件都是一条记录。我使用spark.read.json() 将文件读取为json，没有预定义的模式。我意识到这种存储+读取数据的方式远非理想——镶木地板会更好——但这是我公司atm的现状。
错误检查
- 调用读操作时，spark首先会列出S3中所有底层文件，执行成功。在此之后，它对所有数据进行初始加载，为所有文件构建一个复合 json 模式。正是在这最后一步中发生了错误。
- 我尝试增加/减少内核、执行程序、执行程序内存的数量。 IE。即使有足够多的总执行器内存，错误仍然存在
- 解决这个错误的唯一方法是增加 spark 会话的驱动程序内存，例如10GB+ 内存。但是，一旦我尝试承担更高的数据负载，这种情况就会再次中断。我发现我可以加载的数据量似乎与使用的驱动程序内存量直接相关。
- 我对后一种模式一无所知。在某些情况下，我需要 25GB 的驱动程序内存来完成负载。为什么 spark 需要这么多的驱动程序内存？如果错误确实来自架构构造，为什么/什么
spark.read.json返回给似乎吃掉内存的驱动程序？
代码
import findspark
findspark.init()
import pyspark
spark = (
pyspark.sql.SparkSession.builder
.config('spark.executor.memory', '3g')
.config('spark.executor.cores', '2')
.config('spark.driver.memory','5g')
.config('spark.cores.max', '300')
.getOrCreate()
)
data = spark.read.json('s3a://some-bucket/some-prefix/year=2020/month=01/')
.select('field1', 'field2', 'field3')
完全错误
/opt/spark/python/pyspark/sql/readwriter.py in json(self, path, schema, primitivesAsString, prefersDecimal, allowComments, allowUnquotedFieldNames, allowSingleQuotes, allowNumericLeadingZero, allowBackslashEscapingAnyCharacter, mode, columnNameOfCorruptRecord, dateFormat, timestampFormat, multiLine, allowUnquotedControlChars, lineSep, samplingRatio, dropFieldIfAllNull, encoding)
272 path = [path]
273 if type(path) == list:
--> 274 return self._df(self._jreader.json(self._spark._sc._jvm.PythonUtils.toSeq(path)))
275 elif isinstance(path, RDD):
276 def func(iterator):
/opt/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py in __call__(self, *args)
1255 answer = self.gateway_client.send_command(command)
1256 return_value = get_return_value(
-> 1257 answer, self.gateway_client, self.target_id, self.name)
1258
1259 for temp_arg in temp_args:
/opt/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/opt/spark/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
326 raise Py4JJavaError(
327 "An error occurred while calling {0}{1}{2}.\n".
--> 328 format(target_id, ".", name), value)
329 else:
330 raise Py4JError(
Py4JJavaError: An error occurred while calling o57.json.
: java.lang.OutOfMemoryError: GC overhead limit exceeded
at java.util.Arrays.copyOfRange(Arrays.java:3664)
at java.lang.String.<init>(String.java:207)
at java.lang.String.substring(String.java:1969)
at org.apache.hadoop.fs.Path.<init>(Path.java:219)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$$anonfun$bulkListLeafFiles$3.apply(InMemoryFileIndex.scala:254)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$$anonfun$bulkListLeafFiles$3.apply(InMemoryFileIndex.scala:243)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex$.bulkListLeafFiles(InMemoryFileIndex.scala:243)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.listLeafFiles(InMemoryFileIndex.scala:126)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.refresh0(InMemoryFileIndex.scala:91)
at org.apache.spark.sql.execution.datasources.InMemoryFileIndex.<init>(InMemoryFileIndex.scala:67)
at org.apache.spark.sql.execution.datasources.DataSource.org$apache$spark$sql$execution$datasources$DataSource$$createInMemoryFileIndex(DataSource.scala:533)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:371)
at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.createBaseDataset(JsonDataSource.scala:123)
at org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource$.infer(JsonDataSource.scala:96)
at org.apache.spark.sql.execution.datasources.json.JsonDataSource.inferSchema(JsonDataSource.scala:64)
at org.apache.spark.sql.execution.datasources.json.JsonFileFormat.inferSchema(JsonFileFormat.scala:59)
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$6.apply(DataSource.scala:180)
at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$6.apply(DataSource.scala:180)
at scala.Option.orElse(Option.scala:289)
at org.apache.spark.sql.execution.datasources.DataSource.getOrInferFileFormatSchema(DataSource.scala:179)
at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:373)
at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
at org.apache.spark.sql.DataFrameReader.json(DataFrameReader.scala:391)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
【问题讨论】：
-
您找到确切的原因了吗？我遇到了同样的问题，并将其归结为使用 bulkListLeafFiles 函数的内部并行发现代码导致内存泄漏的事实。但仍然无法找到相同的解决方案。
标签： json apache-spark amazon-s3 pyspark out-of-memory