Dataset.reduce doesn't support shorthand function

Question

I have a simple code:

test("0153") {
  val c = Seq(1,8,4,2,7)
  val max = (x:Int, y:Int)=> if (x > y) x else y
  c.reduce(max)
}

It works fine. But, when I follow the same way to use Dataset.reduce,

test("SparkSQLTest") {
  def max(x: Int, y: Int) = if (x > y) x else y
  val spark = SparkSession.builder().master("local").appName("SparkSQLTest").enableHiveSupport().getOrCreate()
  val ds = spark.range(1, 100).map(_.toInt)
  ds.reduce(max) //compiling error:Error:(20, 15) missing argument list for method max
}

Compiler complains that missing argument list for method max, I don't what's going on here.

ds.reduce((x, y) => max(x,y)) should work – m-bhole Jul 12 '18 at 03:21 — m-bhole, Jul 12 '18 at 03:21
Does `ds.reduce(max _)` work? – Brian McCutchon Jul 12 '18 at 06:33 — Brian McCutchon, Jul 12 '18 at 06:33

score 2 · Accepted Answer · answered Jul 12 '18 at 05:28

2

Change to a function instead of a method and it should work, i.e. instead of

def max(x: Int, y: Int) = if (x > y) x else y

use

val max = (x: Int, y: Int) => if (x > y) x else y

Using the function, using ds.reduce(max) should work directly. More about the differences can be found here.

Otherwise, as hadooper pointed out you can use the method by supplying the arguments,

def max(x: Int, y: Int) = if (x > y) x else y
ds.reduce((x, y) => max(x,y))

answered Jul 12 '18 at 05:28

Shaido

27,497
23
70
73

@hadooper: No problems :) – Shaido Jul 12 '18 at 06:31
1

That‘s interesting, Thanks @Shaido – Tom Jul 12 '18 at 07:16

score 1 · Answer 2 · edited Jul 12 '18 at 05:29

As per spark scala doc, reduce function signature is reduce(func: ReduceFunction[T]): T and reduce(func: (T, T) ⇒ T): T So either of the following will work

Approach 1:

scala> val ds = spark.range(1, 100).map(_.toInt)
ds: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> def max(x: Int, y: Int) = if (x > y) x else y
max: (x: Int, y: Int)Int

scala> ds.reduce((x, y) => max(x,y))
res1: Int = 99

Approach 2 [If you insist on short hand notation like reduce(max)]:

scala> val ds = spark.range(1, 100).map(_.toInt)
ds: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> object max extends org.apache.spark.api.java.function.ReduceFunction[Int]{
     | def call(x:Int, y:Int) = {if (x > y) x else y}
     | }
defined object max

scala> ds.reduce(max)
res3: Int = 99

Hope, this helps!

Thanks @hadooper. The 2nd approach is more like using Java API. For 1st approach, I don't understand why `ds.reduce(max)` doesn't work — Tom, Jul 12 '18 at 04:46

Dataset.reduce doesn't support shorthand function

2 Answers2

Linked