Spark 训练机器学习模型莫名报错java.lang.stackoverflow

遇到一个问题，为此熬了夜，如果没遇到这篇文章，很难发现原因。

具体描述一下问题，我的代码如下：

var dataDf = ...  // load from other place
val inDoubleCols = dataDf.dtypes.filter(
    _._2 != "DoubleType"
).map(_._1).toArray[String]
inDoubleCols.forearch(column => {
    dataDf = dataDf.withColumn(column, col(column).cast(DoubleType))
})

val classifier: LightGBMClassifier = new LightGBMClassifier().setFeaturesCol("features").setLabelCol("label")
val model = classifier.fit(dataDf)

在 fit 那句报错了，如下：

...
21/09/17 11:37:34 INFO SparkContext: Created broadcast 7 from rdd at LightGBMBase.scala:195
21/09/17 11:37:34 INFO FileSourceScanExec: Planning scan with bin packing, max size: 134217728 bytes, open cost is considered as scanning 4194304 bytes.
Exception in thread "main" java.lang.StackOverflowError
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1$adapted(TreeNode.scala:126)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1$adapted(TreeNode.scala:126)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1$adapted(TreeNode.scala:126)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1$adapted(TreeNode.scala:126)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1(TreeNode.scala:126)
	at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$foreachUp$1$adapted(TreeNode.scala:126)
	at scala.collection.immutable.List.foreach(List.scala:431)
	at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
	...

很长很长，实际上就是函数把 stack （栈）给堆满了。我和另外两个前辈没见过这种情况，管这种情况叫“超级树”。

按照一般的思路，在 fit 这行报错，那我们怀疑就是 fit 内部某些机制错了，于是各种更换 scala 版本、 spark 版本、 mmlspark 版本、增加 Xms 等等配置…一直到深夜都没解决“超级树”问题。

当时并不了解 spark 运行机制。那么，问题到底出在哪里呢？

换个思路，其实但凡懂一点计算机原理、思维别太死板也能看出来， 上面的 java.lang.stackoverflow 是因为积攒了太多函数没运行。

为什么到了 fit 这里会积攒太多函数呢？因为 Spark 里面有“懒操作”一说：比如在 数据 dataDf 的 withColumn 这个函数被调用时，不一定要立即去做这件事，而是积攒着，直到 dataDf 需要被缓存、被展示、被计算得到新的值时，之前积攒的一些列操作才开始被调用。 你看我前面 foreach 了那么多 withColumn ，那自然，积攒的函数就很多，就会在 fit 时 java.lang.stackoverflow 。

我当时的解决方案是加入一个 cache() ：

var dataDf = ...  // load from other place
val inDoubleCols = dataDf.dtypes.filter(
    _._2 != "DoubleType"
).map(_._1).toArray[String]
inDoubleCols.forearch(column => {
    dataDf = dataDf.withColumn(column, col(column).cast(DoubleType))
})

dataDf.cache()  // 把之前积攒的懒操作做一做

val classifier: LightGBMClassifier = new LightGBMClassifier().setFeaturesCol("features").setLabelCol("label")
val model = classifier.fit(dataDf)

现在看来，上面的代码还是太糟糕了。

既然都用函数式编程了，就不要有 cols.forearch(c => { x = x... }) 这么愚蠢的写法。而且对每一列进行 withColumn 来转换类型是极其极其低效率的。

如果给我改，该怎么改？

var dataDf = ...  // load from other place
val doubleCols = dataDf.columns.map(
    f => col(f).cast(DoubleType)
)

dataDf.select(doubleCols: _*)
dataDf.show(1, true)  // 把之前积攒的懒操作做一做

val classifier: LightGBMClassifier = new LightGBMClassifier().setFeaturesCol("features").setLabelCol("label")
val model = classifier.fit(dataDf)

注意这里有两个经验，新手必须要越早知道越好：

• select 进行批量类型转换，效率远高于 withColumn
• cache 可以用，但是不能乱用；在不需要把数据传入缓存、只想清空懒操作的情况下， show 没准是更高效的选择

归根结底是当时不了解 Spark 原理。可能小厂就是这样，人手不够，需求还贼多，我连 scala 都没听说过，就接手这么庞大个 Spark 项目，也没有时间学习原理、公司里也没人会没人带我…赶鸭子上架了属于是。现在空下来了，先把原理补一补。

这篇好文章是转载于：学新通技术网