Spark下Dataset[Long]执行reduce强制用java.lang.Long而非scala.Long问题

我有一个单列Dataset[Long]对象x，是通过SparkSession.range方法创建的。按预期我用匿名加法函数_+_对x执行reduce操作应该返回Long类型值，但实际抛出了如下错误：

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.0.0
      /_/
         
Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 11.0.11)
Type in expressions to have them evaluated.
Type :help for more information.

scala> spark
res0: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@a90447f

scala> val x = spark.range(0, 10000000, 10)
x: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> x.reduce(_+_)
<console>:26: error: overloaded method value reduce with alternatives:
  (func: org.apache.spark.api.java.function.ReduceFunction[java.lang.Long])java.lang.Long <and>
  (func: (java.lang.Long, java.lang.Long) => java.lang.Long)java.lang.Long
 cannot be applied to ((java.lang.Long, java.lang.Long) => scala.Long)
       x.reduce(_+_)

就算我明确定义了参数和返回值都是Long类型的加法函数，调用的时候还是报同类错误：

scala> def add(a:Long, b:Long):Long = a+b
add: (a: Long, b: Long)Long

scala> x reduce (add(_,_))
<console>:28: error: overloaded method value reduce with alternatives:
  (func: org.apache.spark.api.java.function.ReduceFunction[java.lang.Long])java.lang.Long <and>
  (func: (java.lang.Long, java.lang.Long) => java.lang.Long)java.lang.Long
 cannot be applied to ((java.lang.Long, java.lang.Long) => scala.Long)
       x reduce (add(_,_))

只有我显式把聚合函数的入参、返回值类型都声明为java.lang.Long的时候，代码才能正常运行：

scala> def add(a:java.lang.Long, b:java.lang.Long):java.lang.Long = a+b
add: (a: Long, b: Long)Long

scala> x.reduce(add(_,_))
res10: Long = 4999995000000

我不觉得这是通用场景下必然会出现的问题，难道在Scala环境里用Spark的Long类型，必须全程用java.lang.Long吗？有没有更简洁的解决方案？

这个报错是Spark Dataset API的双语言兼容设计和Scala的类型匹配机制冲突导致的。SparkSession.range返回的Dataset[Long]，底层编码器默认将Scala的Long映射为Java的java.lang.Long类型，而Dataset的reduce方法为了同时兼容Java和Scala调用，有两个重载实现，编译器匹配重载的时候，无法自动将返回scala.Long的函数映射到接收java.lang.Long的重载方法上，所以报类型不匹配。

不需要全程使用java.lang.Long，可以选择以下更简洁的方案解决问题：

• 方案1：转RDD后调用reduce，RDD的类型系统完全适配Scala原生类型，会自动处理装箱拆箱，写法最简洁：

x.rdd.reduce(_ + _)

• 方案2：使用Spark内置的聚合函数，经过Catalyst优化，性能远高于自定义reduce操作，也不存在类型匹配问题：

import org.apache.spark.sql.functions.sum
val result = x.select(sum("id")).as[Long].head()

• 方案3：调用reduce时显式指定返回类型为java.lang.Long，不需要修改原有函数定义：

// 匿名函数写法
x.reduce((a: Long, b: Long) => a + b : java.lang.Long)
// 自定义add函数的写法
def add(a: Long, b: Long): Long = a + b
x.reduce(add(_: java.lang.Long, _: java.lang.Long))

日常开发完全不需要全程使用java.lang.Long，仅在直接调用Dataset的这类双语言重载方法时才会碰到该问题，优先推荐使用内置SQL函数的方案，性能和代码可读性都更好。

内容的提问来源于stack exchange，提问作者Varun Gawande