[SPARK-21133][CORE] Fix HighlyCompressedMapStatus#writeExternal throws NPE #18343

wangyum · 2017-06-18T07:37:26Z

What changes were proposed in this pull request?

Fix HighlyCompressedMapStatus#writeExternal NPE:

17/06/18 15:00:27 ERROR Utils: Exception encountered
java.lang.NullPointerException
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167)
        at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337)
        at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619)
        at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562)
        at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
17/06/18 15:00:27 ERROR MapOutputTrackerMaster: java.lang.NullPointerException
java.io.IOException: java.lang.NullPointerException
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1310)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167)
        at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337)
        at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619)
        at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562)
        at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.NullPointerException
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303)
        ... 17 more
17/06/18 15:00:27 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 0 to 10.17.47.20:50188
17/06/18 15:00:27 ERROR Utils: Exception encountered
java.lang.NullPointerException
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167)
        at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337)
        at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619)
        at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562)
        at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

How was this patch tested?

manual tests

srowen · 2017-06-18T07:42:09Z

I'm not sure that's a valid fix. This makes this field serialize, when it wasn't intended to. It's either supposed to be recreated on demand, or else, the code needs to deal with it not existing.

wangyum · 2017-06-18T07:47:12Z

cc @jinxing64

SparkQA · 2017-06-18T09:47:47Z

Test build #78224 has finished for PR 18343 at commit 4cf3532.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

jinxing64 · 2017-06-18T14:46:04Z

Thanks for ping.
If I understand correctly, HighlyCompressedStatus is initialized when 2 situations:

Creating MapStatus when shuffle-write and the reduce partitions is over 2000;
Deserialization from stream;

In both situations above, hugeBlockSizes will be initialized.

Could you give some examples how to reproduce this NPE ?

wangyum · 2017-06-18T20:24:13Z

@jinxing64 big_table may be need big enough, my big_table is 270.7 G:

spark-sql -e "
  set spark.sql.shuffle.partitions=2001;
  drop table if exists spark_hcms_npe;
  create table spark_hcms_npe as select id, count(*) from big_table group by id;
"

@srowen I think this issue should be fixed before release 2.2.0, big query throws this exception every time:

SparkQA · 2017-06-18T22:54:18Z

Test build #78230 has finished for PR 18343 at commit 75a9bf1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-06-19T03:20:38Z

Is this still re-producable in current codebase? In the error message above, there is the call to MapOutputTrackerMaster.getSerializedMapOutputStatuses, however, this method is removed in recent change.

viirya · 2017-06-19T04:08:47Z

@wangyum Are you using kyro serializer? I think it is why you hit this issue.

Once you use kyro, I think the readExternal in HighlyCompressedMapStatus won't be used to deserialize the object in driver side. As hugeBlockSizes is a transient variable, it is null now. So when we try to serialize it again, as MapOutputTracker.serializeMapStatuses directly call ObjectOutputStream.writeObject to serialize it, it calls HighlyCompressedMapStatus.writeExternal and cause NPE.

viirya · 2017-06-19T04:22:30Z

I think this should be addressed before 2.2. I already asked notice of other committers on dev mailing list.

wangyum · 2017-06-19T04:31:33Z

@viirya Yes, I' using org.apache.spark.serializer.KryoSerializer, master branch still has this issue, error logs:

17/06/19 12:24:05 ERROR Utils: Exception encountered
java.lang.NullPointerException
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1306)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167)
        at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:728)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:727)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:727)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1340)
        at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:730)
        at org.apache.spark.ShuffleStatus.serializedMapStatus(MapOutputTracker.scala:171)
        at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:389)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
17/06/19 12:24:05 ERROR MapOutputTrackerMaster: java.lang.NullPointerException
java.io.IOException: java.lang.NullPointerException
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1313)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167)
        at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
        at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:728)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:727)
        at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:727)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1340)
        at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:730)
        at org.apache.spark.ShuffleStatus.serializedMapStatus(MapOutputTracker.scala:171)
        at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:389)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.NullPointerException
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1306)
        ... 17 more

viirya · 2017-06-19T04:32:01Z

Because we write/read hugeBlockSizes in writeExternal/readExternal, it seems to me that it is intended to be serialized. So I think removing transient should be ok.

LGTM cc @cloud-fan

viirya · 2017-06-19T04:33:43Z

@wangyum Thanks for updating. Can you try to disable kyro and try it again? So we can verify it.

viirya · 2017-06-19T04:48:20Z

@wangyum Can you also add a test for this?

cloud-fan · 2017-06-19T04:53:58Z

core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala

@@ -141,7 +141,7 @@ private[spark] class HighlyCompressedMapStatus private (
    private[this] var numNonEmptyBlocks: Int,
    private[this] var emptyBlocks: RoaringBitmap,
    private[this] var avgSize: Long,
-    @transient private var hugeBlockSizes: Map[Int, Byte])
+    private[this] var hugeBlockSizes: Map[Int, Byte])


we do want to serialize hugeBlockSizes, but with customized logic, that why we marked it @transient.

I think the corrected fix is, make this class implements KryoSerializable, and copy the customized serialization logic of hugeBlockSizes to kryo serialization hooks.

Sounds good to me. However, the customized serialization logic looks similar to kyro's map serializer. So I'm not sure if it's worth duplicating the customized logic for kyro.

if you can figure out a way to make it serializable with kryo and still keep the customized serialization logic for java serializer, I'm ok with it.

oh seems it is now, then LGTM

SparkQA · 2017-06-19T05:51:53Z

Test build #78239 has finished for PR 18343 at commit e2816ec.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-06-19T05:56:54Z

core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala

@@ -175,6 +175,7 @@ class KryoSerializer(conf: SparkConf)
    kryo.register(None.getClass)
    kryo.register(Nil.getClass)
    kryo.register(Utils.classForName("scala.collection.immutable.$colon$colon"))
+    kryo.register(Utils.classForName("scala.collection.immutable.Map$EmptyMap$"))


why Map$EmptyMap$?

Because this test failed: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/78224/testReport/org.apache.spark.serializer/KryoSerializerSuite/registration_of_HighlyCompressedMapStatus/

SparkQA · 2017-06-19T07:11:59Z

Test build #78248 has finished for PR 18343 at commit facca95.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-06-19T07:25:53Z

core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala

-    @transient private var hugeBlockSizes: Map[Int, Byte])
-  extends MapStatus with Externalizable {
+    @transient private[this] var hugeBlockSizes: Map[Int, Byte])
+  extends MapStatus with Externalizable with KryoSerializable {


I think the previous version already worked... When we have writeExternal and readExternal, @transient doesn't matter for java serializer, so removing @transient to make it work with kryo is a valid fix

OK, I have manual tests remove @transient and Extends KryoSerializable, both worked fine, I'm doing UT.

srowen · 2017-06-19T09:34:20Z

OK I get it. Hm, I wonder why some classes in the code extend Externalizable instead of Serializable? I see a comment about controlling serialization, but Serializable also lets you do that.

I imagine Kryo understands what to do with Serializable but not Externalizable, because i see most classes the implement the latter also implement KryoSerializable.

That seems like the right fix?

Because this seems to just cause default serialization to take over, and that's not desirable, apparently.

If so then we might have a similar problem with CompressedMapStatus, DirectTaskResult, LongHashedRelation, StorageLevel, UpdateBlockInfo, BlockManagerId. It may not make a difference there, or I might misunderstand.

cloud-fan · 2017-06-19T10:52:14Z

It seems Externalizable is kind of abused in Spark, we should benchmark and make sure that these "customized serialization logic" is faster than the default one of java serializer.

For this patch, the kryo serializer seems have the same logic to serialize a map: #18343 (comment) , so it's ok to just remove the @transient

srowen · 2017-06-19T10:53:34Z

I don't quibble with custom serialization logic, but you can do that with Serializable too. And Kryo has its own marker interface too. I wonder what the purpose of Externalizable is then. Actually, I've always wondered this in general about this JDK interface (which extends Serializable) so maybe i'm missing something.

SparkQA · 2017-06-19T11:09:26Z

Test build #78251 has finished for PR 18343 at commit e045bef.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-06-19T14:44:20Z

Test build #78254 has finished for PR 18343 at commit 7a4e6ec.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

rxin · 2017-06-19T17:11:11Z

It's obvious it will reduce data size with custom serialization, since the custom logic doesn't need to write the full classname out which the java default one does.

I don't think Kryo knows what to do with serializable either, unless we specifically register it -- so we'd need to implement both either way whenever we implement custom serialization logic.

vanzin · 2017-06-19T18:05:55Z

since the custom logic doesn't need to write the full classname out

Hmmm... from http://docs.oracle.com/javase/8/docs/api/java/io/Externalizable.html:
"Only the identity of the class of an Externalizable instance is written in the serialization stream..."

So the custom code doesn't need to write the class name because the JRE libs do that for you, so that's not really a win. So it's mostly a cleaner version of the private readObject / writeObject methods.

But re: kryo, yeah, nothing changes.

rxin · 2017-06-19T18:53:13Z

I was talking about the classname for the internal members.

vanzin · 2017-06-19T19:06:04Z

True. I guess since the same serializer instance is not reused, you don't get the benefits of the optimizations that don't require sending the class name after it first shows up..

But back to the change, it feels like the correct thing would be to implement KryoSerializable here to minimize the size of the serialized data when using kryo too. Although maybe that's not too much gain given the class is already registered in KryoSerializer?

vanzin · 2017-06-19T21:37:15Z

I ran a local test and a Map[Int, Byte] with 2000 elements serialized with Kryo ends up at a little less than 14kB. That would be 4 + 4 * 2000 + 1 * 2000 = 9004B using the custom serialization. That's better, but given both numbers are pretty small anyway, just removing the transient looks ok. If it becomes an issue in the future it's pretty easy to add the Kryo interface to the mix.

LGTM.

viirya · 2017-06-20T00:11:48Z

Agreed. The hugeBlockSizes map is not supposed to have too many records but only few huge blocks.

LGTM

…s NPE ## What changes were proposed in this pull request? Fix HighlyCompressedMapStatus#writeExternal NPE: ``` 17/06/18 15:00:27 ERROR Utils: Exception encountered java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) 17/06/18 15:00:27 ERROR MapOutputTrackerMaster: java.lang.NullPointerException java.io.IOException: java.lang.NullPointerException at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1310) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) ... 17 more 17/06/18 15:00:27 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 0 to 10.17.47.20:50188 17/06/18 15:00:27 ERROR Utils: Exception encountered java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) ``` ## How was this patch tested? manual tests Author: Yuming Wang <wgyumg@gmail.com> Closes #18343 from wangyum/SPARK-21133. (cherry picked from commit 9b57cd8) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

cloud-fan · 2017-06-20T01:23:21Z

thanks, merging to master/2.2!

…s NPE ## What changes were proposed in this pull request? Fix HighlyCompressedMapStatus#writeExternal NPE: ``` 17/06/18 15:00:27 ERROR Utils: Exception encountered java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) 17/06/18 15:00:27 ERROR MapOutputTrackerMaster: java.lang.NullPointerException java.io.IOException: java.lang.NullPointerException at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1310) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) ... 17 more 17/06/18 15:00:27 INFO MapOutputTrackerMasterEndpoint: Asked to send map output locations for shuffle 0 to 10.17.47.20:50188 17/06/18 15:00:27 ERROR Utils: Exception encountered java.lang.NullPointerException at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply$mcV$sp(MapStatus.scala:171) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.scheduler.HighlyCompressedMapStatus$$anonfun$writeExternal$2.apply(MapStatus.scala:167) at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303) at org.apache.spark.scheduler.HighlyCompressedMapStatus.writeExternal(MapStatus.scala:167) at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply$mcV$sp(MapOutputTracker.scala:617) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.MapOutputTracker$$anonfun$serializeMapStatuses$1.apply(MapOutputTracker.scala:616) at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1337) at org.apache.spark.MapOutputTracker$.serializeMapStatuses(MapOutputTracker.scala:619) at org.apache.spark.MapOutputTrackerMaster.getSerializedMapOutputStatuses(MapOutputTracker.scala:562) at org.apache.spark.MapOutputTrackerMaster$MessageLoop.run(MapOutputTracker.scala:351) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) ``` ## How was this patch tested? manual tests Author: Yuming Wang <wgyumg@gmail.com> Closes apache#18343 from wangyum/SPARK-21133.

Fix HighlyCompressedMapStatus#writeExternal throws NPE

4cf3532

Kryo register scala.collection.immutable.Map#EmptyMap#

75a9bf1

Remove useless register.

e2816ec

cloud-fan reviewed Jun 19, 2017

View reviewed changes

Extends KryoSerializable

facca95

cloud-fan reviewed Jun 19, 2017

View reviewed changes

Add unit test.

e045bef

Add sc.stop()

7a4e6ec

asfgit closed this in 9b57cd8 Jun 20, 2017

[SPARK-21133][CORE] Fix HighlyCompressedMapStatus#writeExternal throws NPE #18343

[SPARK-21133][CORE] Fix HighlyCompressedMapStatus#writeExternal throws NPE #18343

Conversation

wangyum commented Jun 18, 2017

What changes were proposed in this pull request?

How was this patch tested?

srowen commented Jun 18, 2017

wangyum commented Jun 18, 2017

SparkQA commented Jun 18, 2017

jinxing64 commented Jun 18, 2017

wangyum commented Jun 18, 2017

SparkQA commented Jun 18, 2017

viirya commented Jun 19, 2017

viirya commented Jun 19, 2017

viirya commented Jun 19, 2017

wangyum commented Jun 19, 2017

viirya commented Jun 19, 2017

viirya commented Jun 19, 2017

viirya commented Jun 19, 2017

cloud-fan Jun 19, 2017 • edited

Choose a reason for hiding this comment

viirya Jun 19, 2017 • edited

Choose a reason for hiding this comment

cloud-fan Jun 19, 2017

Choose a reason for hiding this comment

cloud-fan Jun 19, 2017

Choose a reason for hiding this comment

SparkQA commented Jun 19, 2017

cloud-fan Jun 19, 2017

Choose a reason for hiding this comment

wangyum Jun 19, 2017

Choose a reason for hiding this comment

SparkQA commented Jun 19, 2017

cloud-fan Jun 19, 2017

Choose a reason for hiding this comment

wangyum Jun 19, 2017

Choose a reason for hiding this comment

srowen commented Jun 19, 2017

cloud-fan commented Jun 19, 2017

srowen commented Jun 19, 2017

SparkQA commented Jun 19, 2017

SparkQA commented Jun 19, 2017

rxin commented Jun 19, 2017

vanzin commented Jun 19, 2017

rxin commented Jun 19, 2017

vanzin commented Jun 19, 2017

vanzin commented Jun 19, 2017

viirya commented Jun 20, 2017

cloud-fan commented Jun 20, 2017

cloud-fan Jun 19, 2017 •

edited

viirya Jun 19, 2017 •

edited