Scalaの直列化可能なエラー結合が機能しない

Question

2つのRDDの間で結合方法を使用して、それをcassandraに保存しようとしていますが、コードが機能しません。初めに、私は巨大なメイン​​メソッドとすべてうまく動作しますが、私は関数とクラスを使用すると、これは動作しません。私は、Scalaのに新しいですし、

コードがあるスパーク：スレッドの例外「メイン」org.apache.spark.SparkException：タスクで直列化可能ではない私は有名なを取得beginigで

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) 

    } 

。 。 。 私はメインクラスとケースクラスを拡張しますが、stilは機能しません。

Answer 1

ケースクラスをMigrationクラスから専用ファイルおよび/またはオブジェクトに移動する必要があると思います。これはあなたの問題を解決するはずです。さらに、Scalaのケースクラスはデフォルトで直列化可能です。