引子
有一天，一个客户带着下面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性能好，我才用的。。。你不要骗我。。
我：不是用这个还是用那个的问题，是直接不用。。大佬试试嘛
客户：（。。。。。。）神了。。不报错了，速度杠杠滴
我：大佬记住时时刻刻要“避免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的集合类型进行深入的分析
总结与建议
- 首先在我们的Spark程序中，能避免shuffle就尽量避免shuffle
- 当shuffle不可避免的时候，reduceByKey，groupByKey等shuffle算子虽然性能上很直观的可以从字面上可以很容易评价，但是还是有不同的使用场景，用慎重选择
- scala的集合要用好，尽量不要随便的混用，避免一些坑，最好就是好好的研究学习摸索一番
尾声
把测试结果告诉了客户，客户很高心，KPI有望了。。。
我呢，哦，问题又来了。。。
