Scala/Spark serializable Fehler - Join nicht funktionieren

Question

Ich versuche, Join-Methode zwischen 2 RDD zu verwenden und es in Cassandra speichern, aber mein Code nicht funktionieren. Am Anfang bekomme ich eine große Main-Methode und alles funktioniert gut, aber wenn ich Funktion und Klasse verwende, funktioniert das nicht. Ich bin neu zu scala und Funken

Code ist:

class Migration extends Serializable { 

    case class userId(offerFamily: String, bp: String, pdl: String) extends Serializable 
    case class siteExternalId(site_external_id: Option[String]) extends Serializable 
    case class profileData(begin_ts: Option[Long], Source: Option[String]) extends Serializable 

    def SparkMigrationProfile(sc: SparkContext) = { 

    val test = sc.cassandraTable[siteExternalId](KEYSPACE,TABLE) 
    .keyBy[userId] 
    .filter(x => x._2.site_external_id != None) 

    val profileRDD = sc.cassandraTable[profileData](KEYSPACE,TABLE) 
    .keyBy[userId] 

    //dont work 
    test.join(profileRDD) 
    .foreach(println) 

    // don't work 
    test.join(profileRDD) 
    .saveToCassandra(keyspace, table) 

    } 

Am beginig i die berühmten erhalten: Exception in thread "main" org.apache.spark.SparkException: Aufgabe nicht serializable an. . . also erweitere ich meine Hauptklasse und auch die Fallklasse aber trotzdem funktioniert das nicht.

Answer 1

Ich denke, Sie sollten Ihre Fallklassen von Migration Klasse in dedizierte Datei und/oder Objekt verschieben. Dies sollte dein Problem lösen. Darüber hinaus sind Scala-Fallklassen standardmäßig serialisierbar.