Изменение типа данных var в Spark с использованием его собственного метода

Question

Ниже приведен минимальный пример проблемы, с которой я сталкиваюсь. У меня есть массив, который я хочу изменить на месте, так как он содержит около миллиона элементов. следующий код работает, за исключением самого последнего оператора.

 import spark.implicits._

 case class Frame(x: Double, var y: Array[Double]) {
    def total(): Double = {
        return y.sum
    }
    def modifier(): Unit = {
        for (i <- 0 until y.length) {
            y(i) += 10
        }
        return
    }
 }

 val df = Seq(
               (1.0, Array(0, 2, 1)),
               (8.0, Array(1, 2, 3)),
               (9.0, Array(11, 21, 23))
             ).toDF("x", "y")

 val ds = df.as[Frame]
 ds.show

 ds.map(_.total()).show     // works
 ds.map(_.modifier()).show  // does not work

Ошибка выглядит следующим образом:

scala> ds.map(_.modifier()).show
<console>:50: error: Unable to find encoder for type Unit. An implicit Encoder[Unit] is needed to store Unit instances in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.
       ds.map(_.modifier()).show

Я не вижу причину проблемы. Буду благодарен за любую помощь в исправлении ошибки.

Answer 1

На самом деле, это не имеет ничего общего с 'var' или 'val', оно касается изменяемых структур данных. Проблема в том, что modifier возвращает Unit (например, ничего), поэтому вы не можете отобразить эти результаты. Вы можете запустить его, используя:

case class Frame(x: Double, var y: Array[Double]) {
def total(): Double = {
  return y.sum
}
def modifier(): Frame = {
  for (i <- 0 until y.length) {
    y(i) += 10
  }
  return this
}

}

Но, по моему мнению, нет особого смысла, вам следует избегать изменяемого состояния. Кроме того, я бы хотел, чтобы классы дел были простыми (т.е. без logi c) в spark, использовали их только в качестве контейнеров данных. Если вам нужно увеличить каждый элемент к тому времени, вы можете сделать это также так:

case class Frame(x: Double, val y: Array[Double])

ds.map(fr => fr.copy(y = fr.y.map(_+10.0))).show