rdd.map中引用外部val报错"Task not serializable"

Question

  
val Size=5
val fixedLogs = rdd.map{row =>{
    Size
  }
  }在spark-shell 中声明变量val Size=5；然后在map中引用，报错"Task not serializable"该如何处理谢谢大神们

Answer 1

val bcSize = sc.broadCast(size)
val fixedLogs = rdd.map{row =>{
    bcSize.value
  }
  }

Answer 2

因为你RDD的transform和action算子是在executor上执行的，而你写的main方法，是在driver上运行，所以你不可能跨机器跨进程去访问你在driver上的size。要访问，只能用广播变量。这个是在每一个executor上产生一个被广播对象的副本。

Answer 3

我用的是SQL的dataFrame.rdd.map
以下代码还是不能序列化val bcSize = sc.broadcast(5)
val test = dayErrorDF.select("judge.*","fix.*").rdd.map{row =>bcSize.value
}

Answer 4

scala> val df = sqlContext.range(1,1000)
df: org.apache.spark.sql.DataFrame = [id: bigint]scala> val bcSize = sc.broadcast(5)
bcSize: org.apache.spark.broadcast.Broadcast[Int] = Broadcast(0)scala> val m = df.rdd.map{ row => bcSize.value }
m: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at map at <console>:31scala> m.count
res2: Long = 999一点问题没有啊

Answer 5

dayErrorDF.select("judge.*","fix.*") 我觉得是这部分的问题，我没见过这种写法。
你单独dayErrorDF.select("judge.*","fix.*").show看看会不会报错

Answer 6

dayErrorDF.select("judge.*","fix.*") 我觉得是这部分的问题，我没见过这种写法。
你单独dayErrorDF.select("judge.*","fix.*").show看看会不会报错可以显示，就是一个嵌套，把“.*”去掉也不行，很奇怪。

Answer 7

dayErrorDF.select("judge.*","fix.*") 我觉得是这部分的问题，我没见过这种写法。
你单独dayErrorDF.select("judge.*","fix.*").show看看会不会报错可以显示，就是一个嵌套，把“.*”去掉也不行，很奇怪。
你这问题刷新了我的三观spark版本是多少？

Answer 8

dayErrorDF.select("judge.*","fix.*") 我觉得是这部分的问题，我没见过这种写法。
你单独dayErrorDF.select("judge.*","fix.*").show看看会不会报错可以显示，就是一个嵌套，把“.*”去掉也不行，很奇怪。
你这问题刷新了我的三观spark版本是多少？1.6.0