Класс Case в foreachRDD вызывает ошибку сериализации - PullRequest
Новая
       25

Класс Case в foreachRDD вызывает ошибку сериализации

0 голосов
/ 06 января 2019

Я могу создать DF внутри foreachRDD, если я не попытаюсь использовать Класс Case и просто позволю делать имена столбцов по умолчанию с помощью toDF () или если я назначу их через toDF ("c1," c2 ").

Как только я пытаюсь использовать Case Class и, посмотрев на примеры, я получаю: 

Task not serializable

Если я переверну оператор Case Class вокруг, я получу: 

toDF() not part of RDD[CaseClass]

Это унаследовано, но мне любопытно относительно n-й ошибки Сериализации, которую Spark может произвести и если она переносится в Структурированную Потоковую передачу.

У меня есть СДР, который не нужно разбивать, может в этом проблема? NO. Запуск в DataBricks?

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

import org.apache.spark.sql.SparkSession
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable

case class Person(name: String, age: Int) //extends Serializable // Some say inherently serializable so not required

val spark = SparkSession.builder
    .master("local[4]")
    .config("spark.driver.cores", 2)
    .appName("forEachRDD")
    .getOrCreate()

val sc = spark.sparkContext
val ssc = new StreamingContext(spark.sparkContext, Seconds(1)) 

val rddQueue = new mutable.Queue[RDD[List[(String, Int)]]]()
val QS = ssc.queueStream(rddQueue) 

QS.foreachRDD(q => {
   if(!q.isEmpty) {   
      import spark.implicits._
      val q_flatMap = q.flatMap{x=>x}
      val q_withPerson = q_flatMap.map(field => Person(field._1, field._2))
      val df = q_withPerson.toDF() 
      df.show(false)
   }
 }
)

ssc.start()
for (c <- List(List(("Fred",53), ("John",22), ("Mary",76)), List(("Bob",54), ("Johnny",92), ("Margaret",15)), List(("Alfred",21), ("Patsy",34), ("Sylvester",7)) )) {
 rddQueue += ssc.sparkContext.parallelize(List(c))
} 
ssc.awaitTermination()

1 Ответ

0 голосов
/ 06 января 2019

Не вырос на Java, но, посмотрев вокруг, я понял, что делать, но не настолько опытен, чтобы объяснять.

Я работал в блокноте DataBricks, где создавал прототип.

Подсказка в том, что 

case class Person(name: String, age: Int)

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

...