Spark Shell：任务不可序列化

Question

我对Spark，Scala和Cassandra都是新手。使用Spark，我试图从MySQL获取一些ID。

import org.apache.spark.rdd.JdbcRDD
import java.sql.{Connection, DriverManager, ResultSet}
Class.forName("com.mysql.jdbc.Driver").newInstance

import com.datastax.spark.connector._, org.apache.spark.SparkContext, org.apache.spark.SparkContext._, org.apache.spark.SparkConf

val myRDD = new JdbcRDD( sc, () => DriverManager.getConnection(url,username,password) ,"select id from user limit ?, ?",1, 20, 10, r => r.getString("id")) ;
myRDD.foreach(println) 

我可以看到在控制台打印的ID。

object HelloWorld { 
       def sum(id : String): Unit = {
        val each_spark_rdd = uplink_rdd.select("number").where("id=?",Id).as((c: Int) => c).sum
        println(each_spark_rdd)
      }
  }

 val uplink_rdd = sc.cassandraTable("keyspace", "table")

scala> HelloWorld.sum("5") 
50

当我试图在每个提取id上运行相同的函数时

myRDD.map(HelloWorld.sum)
or
myRDD.foreach(HelloWorld.sum)
or 
for (id <- myRDD) HelloWorld.sum(id)

它给出与例外相同的例外

在阅读spark-shell中的Apache spark：“sparkException：Task not serializable”后，我尝试将@transient添加到RDDs中

@transient val myRDD = new JdbcRDD ...
@transient val uplink_rdd = sc.cassandra....

Answer 1

您的代码试图在myrdd的转换中使用uplink_rdd。应用于RDD的闭包不能包含另一个RDD。

您应该按照joinwithcassandratable的方式来做一些事情，它将并行和分布式（ly？）使用myrdd中的信息从Cassandra中提取数据。如果从Cassandra中提取单个分区键，则此操作有效

查看文档

val cc = CassandraConnector(sc.getConf)
myRDD.mapPartitions { it =>
  cc.withSessionDo { session =>
    session.execute("whatever query you want")
  }
}

类似于

myRDD.collect.foreach(HelloWorld.sum)