Задача не сериализуема, когда класс сериализуем

Question

У меня есть следующий класс в Scala

case class A
  (a:Int,b:Int) extends Serializable

, когда я пробую в Spark 2.4. (через Databricks)

val textFile = sc.textFile(...) 
val df = textFile.map(_=>new A(2,3)).toDF()

(Изменить: ошибка возникает, когда я вызываю df.collect () или регистрирую как таблицу)

Я получаю org.apache.spark.SparkException: Task not serializable

что мне не хватает?

Я пробовал добавлять кодировщики:

implicit def AEncoder: org.apache.spark.sql.Encoder[A] = 
  org.apache.spark.sql.Encoders.kryo[A]

и

import spark.implicits._
import org.apache.spark.sql.Encoders

редактировать: Я также пробовал:

val df = textFile.map(_=>new A(2,3)).collect()

но пока не повезло.

Answer 1

Иногда это происходит с перерывами на DataBricks. Очень раздражает.

Перезагрузите кластер и попробуйте еще раз, у меня иногда возникала эта ошибка, но после перезапуска она не возникала.

Answer 2

Вы можете напрямую проанализировать файл как Dataset с имеющимся у вас классом case.

case class A(a:Int,b:Int) extends Serializable
val testRDD = spark.sparkContext.textFile("file:///test_file.csv")
val testDS = testRDD.map( line => line.split(",")).map(line_cols => A(line_cols(0).toInt, line_cols(1).toInt) ).toDS()

#res23: org.apache.spark.sql.Dataset[A] = [a: int, b: int]