Spark NullPointerException： 无法调用 invalidateSerializedMapOutputStatusCache()，因为 "shuffleStatus" 为 nul

Question

我正在使用Java 17和UDF在Windows 10上运行一个简单的小型Spark 3.3.0管道。我几乎没有做任何有趣的事情，现在当我只在30000条记录上运行管道时，我得到的是：

[ERROR] Error in removing shuffle 2
java.lang.NullPointerException: Cannot invoke "org.apache.spark.ShuffleStatus.invalidateSerializedMapOutputStatusCache()" because "shuffleStatus" is null
at org.apache.spark.MapOutputTrackerMaster.$anonfun$unregisterShuffle$1(MapOutputTracker.scala:882)
at org.apache.spark.MapOutputTrackerMaster.$anonfun$unregisterShuffle$1$adapted(MapOutputTracker.scala:881)
at scala.Option.foreach(Option.scala:437)
at org.apache.spark.MapOutputTrackerMaster.unregisterShuffle(MapOutputTracker.scala:881)
at org.apache.spark.storage.BlockManagerStorageEndpoint$$anonfun$receiveAndReply$1.$anonfun$applyOrElse$3(BlockManagerStorageEndpoint.scala:59)
at scala.runtime.java8.JFunction0$mcZ$sp.apply(JFunction0$mcZ$sp.scala:17)
at org.apache.spark.storage.BlockManagerStorageEndpoint.$anonfun$doAsync$1(BlockManagerStorageEndpoint.scala:89)
at scala.concurrent.Future$.$anonfun$apply$1(Future.scala:678)
at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:467)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
at java.base/java.lang.Thread.run(Thread.java:833)

我搜索了一下，在错误消息中找不到任何主要术语。

令人不安的是，Spark正在突破似乎是处理的一个基本部分，并在其中使用NullPointerException。

我提交了机票SPARK-40582。

Answer 1

我提交了SPARK-40582，他们告诉我这是一个已知的Scala 2.13.8问题(#12613(。他们在SPARK-39553中添加了一个修复程序，计划在v3.3.1中发布。

Answer 2

好的，我不知道Spark，但我参考了下面的两页。

https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/MapOutputTracker.scalaunregisterShuffle方法

https://www.hadoopinrealworld.com/how-does-shuffle-sort-merge-join-work-in-spark/

Spark对数据进行混洗，在混洗过程中，找不到一些根据id应该存在的数据。

选项1：试着增加分配给应用程序的内存，看看这是否能解决这个问题。

选项2：对各种用例进行单元测试，看看当你遇到问题时是否能找到用例。

选项3：尝试早期版本的spark。

根据OP评论更新了我的建议理由-我同意你的观点，我的回答似乎是"；重新启动计算机"；一般性的建议。以下是我给出上述3条建议的具体原因。

1. 我查找了发生NullPointerException的spark的源代码(给定链接(。我看到一个ConcurrentHashMap shuffleStatus，其中需要shuffleId的值，但找不到。你现在编辑的问题提到你是非决定性地得到错误的。也就是说，对于相同的数据，您会出现一次错误，但下次不会。这指出了问题的非代码原因，比如内存可用性——因此建议#1

2. 解决这种"非确定性"错误的一般方法是使用不同的可能用例对代码进行单元测试。这有助于您在出现此类错误时确定用例。因此建议#2。

3. 库代码内部出现类似NullPointerException的运行时异常表明某些代码流未得到处理，并且库代码存在错误。新版本的库可能会出现这样的错误。在这种情况下，谨慎的做法是向github社区报告此类错误。我们可以使用更早、更稳定的版本来解锁自己——因此建议#3。(我想在我写原始答案时，你还没有报告这个错误(