我在 Spark 2.4.4 中使用带有大窗口的窗口函数,例如。
Window
.partitionBy("id")
.orderBy("timestamp")
在我的测试中,我有大约 70 个不同的 ID,但我可能有大约 200 000 行 ID。如果没有进一步的配置,我必须为我的执行器分配大量内存以避免这种 OOM:
org.apache.spark.memory.SparkOutOfMemoryError: Unable to acquire 16384 bytes of memory, got 0
at org.apache.spark.memory.MemoryConsumer.throwOom(MemoryConsumer.java:157)
at org.apache.spark.memory.MemoryConsumer.allocateArray(MemoryConsumer.java:98)
at org.apache.spark.util.collection.unsafe.sort.UnsafeInMemorySorter.<init>(UnsafeInMemorySorter.java:128)
at org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.<init>(UnsafeExternalSorter.java:161)
at org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter.create(UnsafeExternalSorter.java:128)
at org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray.add(ExternalAppendOnlyUnsafeRowArray.scala:115)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11$$anon$1.fetchNextPartition(WindowExec.scala:345)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11$$anon$1.next(WindowExec.scala:371)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11$$anon$1.next(WindowExec.scala:303)
at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage15.processNext(Unknown Source)
at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$12$$anon$1.hasNext(WholeStageCodegenExec.scala:631)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11$$anon$1.fetchNextRow(WindowExec.scala:314)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11$$anon$1.<init>(WindowExec.scala:323)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11.apply(WindowExec.scala:303)
at org.apache.spark.sql.execution.window.WindowExec$$anonfun$11.apply(WindowExec.scala:302)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:801)
at org.apache.spark.rdd.RDD$$anonfun$mapPartitions$1$$anonfun$apply$23.apply(RDD.scala:801)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:49)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
查看源代码,我发现了这个参数,根本没有记录:
spark.sql.windowExec.buffer.in.memory.threshold
给它一个大的尺寸(例如1.000.000),我不再需要那么多的内存。据我了解,这是缓冲的行数;我想增加这个参数不会重复执行程序内存中的行,但这对我来说并不是很清楚。
有人可以准确地解释一下执行器端如何处理窗口吗?为什么数据会重复?如何避免这种重复并使过程更快,每个窗口中有许多行?可以使用哪些参数?
Thx.
我发现了这个参数,它根本没有记录:
它是一个内部配置属性。
通过阅读源代码,我设法“收集”以下内容:
Spark.sql.windowExec.buffer.in.memory.threshold(内部)保证在内存中保存的行数阈值WindowExec
物理操作员。
默认:4096
Use SQLConf.windowExecBufferInMemoryThreshold
方法来访问当前的
价值。
说到内部属性WindowExec
操作员,您可能还需要另一个来进行性能调整:
Spark.sql.windowExec.buffer.spill.threshold(内部)缓冲行数的阈值WindowExec
物理操作员。
默认:4096
Use SQLConf.windowExecBufferSpillThreshold
方法来访问当前值。
唉,我无法完全解释其内部原理。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)