一个引发Spark在Executor端Shuffle Write

引子

有一天，一个客户带着下面StackOverFlow错误栈来找我，简单的主诉之后发现应该是shuffle wtrite阶段写中间文件的时候序列化发生了死循环。

java.lang.StackOverflowError
    at net.jpountz.util.SafeUtils.checkLength(SafeUtils.java:37)
    at net.jpountz.util.SafeUtils.checkRange(SafeUtils.java:29)
    at net.jpountz.lz4.LZ4BlockOutputStream.write(LZ4BlockOutputStream.java:151)
    at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1876)
    at java.io.ObjectOutputStream$BlockDataOutputStream.write(ObjectOutputStream.java:1840)

客户：xss10m了，还是StackOverFlow，Spark好坑。。
我：莫不是lz4有bug，大佬，试试snappy
客户：(。。。。。。) 不行

ava.lang.StackOverflowError
        at org.xerial.snappy.SnappyOutputStream.write(SnappyOutputStream.java:123)
        at org.apache.spark.io.SnappyOutputStreamWrapper.write(CompressionCodec.scala:201)
        at java.io.ObjectOutputStream$BlockDataOutputStream.drain(ObjectOutputStream.java:1876

我：大佬要不试试不压缩 spark.shuffle.spill.compress false
客户：（。。。。。。）不行

java.lang.StackOverflowError
    at sun.misc.Unsafe.getLong(Native Method)
    at java.io.ObjectStreamClass$FieldReflector.getPrimFieldValues(ObjectStreamClass.java:1966)
    at java.io.ObjectStreamClass.getPrimFieldValues(ObjectStreamClass.java:1233)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1532)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)

我：你应用代码中是不是有定义新的类，但没有注册给Kyro
客户：是有
我：大佬试试？
客户：（。。。。。。）不行，心酸

java.lang.StackOverflowError
    at com.esotericsoftware.kryo.serializers.FieldSerializer$CachedField.<init>(FieldSerializer.java:689)
    at com.esotericsoftware.kryo.serializers.ObjectField.<init>(ObjectField.java:43)
    at com.esotericsoftware.kryo.serializers.UnsafeCacheFields$UnsafeObjectField.<init>(UnsafeCacheFields.java:295)

我：（这位大佬的耐心应该已近没有了，10086的技能用的差不多了）大佬还是把源码与我一阅吧。
客户：好。

// 客户源码占位符

测试1

拿到用户代码一看，虽然写的烂是烂了点，但这么短小应该不至于因为函数链太长二导致StackOverFlow。

简单模拟下，客户的场景

sc.range(1, 1000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, scala.collection.mutable.LinkedList(2, 4, 6))).reduceByKey( (x, y) => x ++ y).filter(p => p._1 == 1).flatMap(p => p._2).take(1)

我：（肉眼一看，兄弟你这reduceByKey要来作甚。。。）大佬，您的case实际不需要，把reduceByKey去掉吧。。
客户：官文说reduceByKey性能好，我才用的。。。你不要骗我。。

315c9904576a98f96d5c6468026990b7.png

我：不是用这个还是用那个的问题，是直接不用。。大佬试试嘛
客户：（。。。。。。）神了。。不报错了，速度杠杠滴
我：大佬记住时时刻刻要“避免shuffle”啊。
客户：谢谢。

测试2

等等，今天把他shuffle搞掉了，改天再来个避无可避的怎么办？

还是等搞搞清楚

case 0

用户case的变种，模拟StackOverFlow

sc.range(1, 100000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, scala.collection.mutable.LinkedList(2, 4, 6))).reduceByKey( (x, y) => x ++ y).filter(p => p._1 == 1).flatMap(p => p._2).take(1)

直接用bin/spark-shell --master yarn运行

spark.executor.instances=6
spark.executor.memory=5g
spark.executor.cores=1

任务失败，如期而至的StackOverFlow

Job aborted due to stage failure: Task 5 in stage 0.0 failed 4 times, most recent failure: Lost task 5.3 in stage 0.0 (TID 16, hzadg-hadoop-dev3.server.163.org, executor 9): java.lang.StackOverflowError +details
Job aborted due to stage failure: Task 5 in stage 0.0 failed 4 times, most recent failure: Lost task 5.3 in stage 0.0 (TID 16, hzadg-hadoop-dev3.server.163.org, executor 9): java.lang.StackOverflowError
    at java.lang.ref.WeakReference.<init>(WeakReference.java:69)
    at java.io.ObjectStreamClass$WeakClassKey.<init>(ObjectStreamClass.java:2306)
    at java.io.ObjectStreamClass.lookup(ObjectStreamClass.java:322)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1133)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
    at 
....
....
....
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)

case 1

先把reduceByKey改成groupByKey试试看，是不是只要有shuffle就会有问题，要是那可能就是spark带的bug或者限制了。

sc.range(1, 100000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, scala.collection.mutable.LinkedList(2, 4, 6))).groupByKey(10).filter(p => p._1 == 1).flatMap(p => p._2).take(1)
res2: Array[Seq[Int]] = Array(LinkedList(2, 4, 6))

执行成功，所以Spark自带问题的可能性极小

case 2

那是不是数据太大的问题呢，再把数据量改小试试

scala> sc.range(1, 1000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, scala.collection.mutable.LinkedList(2, 4, 6))).reduceByKey( (x, y) => x ++ y).filter(p => p._1 == 1).flatMap(p => p._2).take(1)
res3: Array[Int] = Array(2)

执行成功，说明数据量的确会触发这个问题，看来是reduceByKey把Value的List Combine的过大了，但是数据不是存在堆里的吗，怎么也应该以是OOM而不是StackOverFlow嘛。。

case 3

带着一丝疑问，以及最初对客户代码烂的疑惑，我们把LinkedList改成Seq，使得map的output对象一致试试，数据量还是保持在之前出错的级别

scala> sc.range(1, 100000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, Seq(2, 4, 6))).reduceByKey( (x, y) => x ++ y).filter(p => p._1 == 1).flatMap(p => p._2).take(1)
res4: Array[Int] = Array(2)

执行成功

再增大一个数量级？

scala> sc.range(1, 1000000, 5).map(x => if (x % 2 == 1) (0, Seq(1, 2, 3)) else (1, Seq(2, 4, 6))).reduceByKey( (x, y) => x ++ y).filter(p => p._1 == 1).flatMap(p => p._2).take(1)
res5: Array[Int] = Array(2)

还是成功了，看来这个锅是scala的集合的咯，在大数量下，可变集合和不可变集合反复连接后，序列化貌似触发scala内核的bug

// TODO 对scala的集合类型进行深入的分析

总结与建议

1. 首先在我们的Spark程序中，能避免shuffle就尽量避免shuffle
2. 当shuffle不可避免的时候，reduceByKey，groupByKey等shuffle算子虽然性能上很直观的可以从字面上可以很容易评价，但是还是有不同的使用场景，用慎重选择
3. scala的集合要用好，尽量不要随便的混用，避免一些坑，最好就是好好的研究学习摸索一番

尾声

把测试结果告诉了客户，客户很高心，KPI有望了。。。

我呢，哦，问题又来了。。。