Spark踩坑填坑-聚合函数-序列化异常
部分内容原文地址：
AISeekOnline：Spark Caused by: java.io.NotSerializableException 序列化异常踩过的坑
zhou_jun：Spark运行程序异常信息： org.apache.spark.SparkException: Task not serializable 解决办法
一、Spark聚合函数特殊场景
在对数据进行统计分析时，如果对指标进行聚合运算，而待查询的字段中还包含了维度，则原则上我们还需要按照维度字段进行分组。倘若这个聚合运算为sum函数，分组之后就相当于分类汇总了。有一种特殊场景是我们对指标执行了sum聚合，查询字段也包含了维度，但我们不希望对维度分组。例如：
select name, role, sum(income) from employee
虽然返回的结果挺奇怪，因为它事实上是针对整张表的income进行了求和运算，与name、role无关。查询结果中返回的其实是第一条记录的name与role。但至少在MySQL中，这样的SQL语法是正确的。
但是在Spark中，执行这样的SQL语句，则会抛出org.apache.spark.sql.AnalysisException异常：
org.apache.spark.sql.AnalysisException: expression 'employee.`name`' is neither present in the group by, nor is it an aggregate function. Add to group by or wrap in first() (or first_value) if you don't care which value you get.
这是因为Spark SQL在执行SQL语句时，事先会调用CheckAnalysis的checkAnalysis方法对LogicalPlan进行检查：
def checkAnalysis(plan: LogicalPlan): Unit = {
case e: Attribute if groupingExprs.isEmpty =>
// Collect all [[AggregateExpressions]]s.
val aggExprs = aggregateExprs.filter(_.collect {
case a: AggregateExpression => a
}.nonEmpty)
failAnalysis(
s"grouping expressions sequence is empty, " +
s"and '${e.sql}' is not an aggregate function. " +
s"Wrap '${aggExprs.map(_.sql).mkString("(", ", ", ")")}' in windowing " +
s"function(s) or wrap '${e.sql}' in first() (or first_value) " +
s"if you don't care which value you get."
)
case e: Attribute if !groupingExprs.exists(_.semanticEquals(e)) =>
failAnalysis(
s"expression '${e.sql}' is neither present in the group by, " +
s"nor is it an aggregate function. " +
"Add to group by or wrap in first() (or first_value) if you don't care " +
"which value you get.")
}
按照给出的SQL语句，groupingExprs应该是Empty才对，然而根据抛出的错误提示，在对分析语句进行检查时，却是走的后一个模式匹配分支，即e: Attribute if !groupingExprs.exists(_.semanticEquals(e))。
根据提示，在不增加group by的情况下，需要对select中的字段包裹一个first()或者first_value()函数，如下所示：
spark.sql("select first(name),first(role), sum(income) from employee")
这里的维度包含name和role。如果添加了group by，但只针对其中的一个维度进行了分组，那么对于缺少分组的维度，也当用first()函数来包裹才对。
第一部分内容转载自：张逸-简书
二、spark sql group by
hiveContext.sql("select time,count(*) from page_click group by id").collect.foreach(println)
//报错
org.apache.spark.sql.AnalysisException: expression 'page_click.`time`' is neither present in the group by, nor is it an aggregate function. Add to group by or wrap in first() (or first_value) if you don't care which value you get.;
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:40)
at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:58)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.org$apache$spark$sql$catalyst$analysis$CheckAnalysis$class$$anonfun$$checkValidAggregateExpression$1(CheckAnalysis.scala:218)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$6.apply(CheckAnalysis.scala:245)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1$$anonfun$apply$6.apply(CheckAnalysis.scala:245)
at scala.collection.immutable.List.foreach(List.scala:381)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:245)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:67)
at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:126)
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.checkAnalysis(CheckAnalysis.scala:67)
at org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyzer.scala:58)
at org.apache.spark.sql.execution.QueryExecution.assertAnalyzed(QueryExecution.scala:49)
at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:64)
at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:582)
at org.apache.spark.sql.SQLContext.sql(SQLContext.scala:682)
... 48 elided
time这一列在 group by的时候有多个查询结果，需要使用collect_set()一下。
scala>hiveContext.sql("select collect_set(time),count(*) from page_click group by id limit 10").collect.foreach(println)
第二部分内容转载自：time_exceed-CSDN
三、Spark Caused by: java.io.NotSerializableException 序列化异常踩过的坑
这部分之所以会进行转载学习，是因为在之前的项目中，DateTimeFormat这个方法和Redis初始化对象方法，会报Spark Caused by: java.io.NotSerializableException 这个错，故转载过来学习。
最近有需求需要在driver端创建好类实例，然后在rdd里面调用，但是使用过程中发现 Caused by: java.io.NotSerializableException，即序列化异常，通过查处网上资料发现是构建的类没有继承Serializable，没有继承Serializable的类是不会自动执行自动序列化操作的，因此我把构建的类继承了Serializable这个类，再次运行的时候发现依旧是序列化异常，百般不得其解。
序列异常天坑1（网上常见的）
在rdd外实例化的类没有继承Serializable，在实例化类在rdd中使用，如下代码块：
class ClassA {
def getClassName: String = this.getClass.getSimpleName
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classA = new ClassA()
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in main -> " + classA.getClassName + s": $i")
.collect().foreach(println)
}
}
填坑方法
方法1：将ClassA修改为继承Serializable类：
class ClassA extends Serializable {
def getClassName: String = this.getClass.getSimpleName
}
方法2：将ClassA放在rdd里面进行实例化：
rdd.map(i => {
val classA = new ClassA
"getClassName in main -> " + classA.getClassName + s": $i"
}).collect().foreach(println)
方法3：将ClassA改成静态类，静态类自动实例化，在rdd里面直接调用其方法：
object ClassA {
def getClassName: String = this.getClass.getSimpleName
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in main -> " + ClassA.getClassName + s": $i")
.collect().foreach(println)
}
}
序列异常天坑2
在rdd里面调用类中某个类的方法报序列化异常，代码如下：
class ClassA {
def getClassName: String = this.getClass.getSimpleName
}
class ClassB(sc: SparkContext) extends Serializable{
val classA = new ClassA()
def fun(): Unit = {
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in ClassB -> "+classA.getClassName + s": $i")
.collect.foreach(println)
}
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classB = new ClassB(sc)
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in main -> " + classB.classA.getClassName + s": $i")
.collect().foreach(println)
}
}
如上述，在rdd里面调用ClassB中属性ClassA中的方法报序列化异常
填坑方法
方法1：这个ClassB有点脑残，把ClassA作为属性实在不可取，如果只是为了达到调用ClassA内的方法，则可以让ClassB去继承ClassA
class ClassA extends Serializable {
def getClassName: String = this.getClass.getSimpleName
}
class ClassB(sc: SparkContext) extends ClassA with Serializable{
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classB = new ClassB(sc)
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in main -> " + classB.getClassName + s": $i")
.collect().foreach(println)
}
}
方法2：在rdd外先把ClassB中ClassA取出来放到一个变量里面去，再在rdd里面调用该变量：
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classB = new ClassB(sc)
val a = classB.classA
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in main -> " + a.getClassName + s": $i")
.collect().foreach(println)
}
}
这种类似填坑1里面的，相当于重新new了一个ClassA
序列异常天坑3
在类ClassB中有方法fun，和属性classA，fun调用了classA中的方法：
class ClassA extends Serializable {
def getClassName: String = this.getClass.getSimpleName
}
class ClassB(sc: SparkContext) extends Serializable{
val classA = new ClassA()
def fun(): Unit = {
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in ClassB -> "+classA.getClassName + s": $i")
.collect.foreach(println)
}
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classB = new ClassB(sc)
classB.fun()
}
}
填坑方法
方法1：在fun里面不使用属性classA，而是在fun里面重新构建ClassA
def fun(): Unit = {
val classA = new ClassA()
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in ClassB -> "+classA.getClassName + s": $i")
.collect.foreach(println)
}
这类似于天坑1的解决方式。但是很多时候我们的ClassA是一个比较全的工具类，不仅仅是在fun单个方法体里面调用，因此需要将放到ClassB作为属性。
方法2：与前面的一样，可以在fun方法里面的rdd前面先新增一个变量在调用
def fun(): Unit = {
val a = classA
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in ClassB -> "+a.getClassName + s": $i")
.collect.foreach(println)
}
方法3：把ClassB修改成object修饰静态类：
class ClassA extends Serializable {
def getClassName: String = this.getClass.getSimpleName
}
object ClassB extends Serializable{
val classA = new ClassA()
def fun(sc: SparkContext): Unit = {
val rdd = sc.makeRDD(1 to 5)
rdd.map(i => "getClassName in ClassB -> "+classA.getClassName + s": $i")
.collect.foreach(println)
}
}
object SerializableTest {
def main(args: Array[String]): Unit = {
val conf = new SparkConf(true)
.setMaster("local[*]")
.setAppName("SerializableTest")
.set("spark.rdd.compress", "true")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
lazy val sc = new SparkContext(conf)
val classB = ClassB
classB.fun(sc)
}
}
四、org.apache.spark.SparkException: Task not serializable
错误信息：
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
问题原因：再对RDD进行操作时引用了类的成员变量而该成员变量无法被序列化所导致的。
例如：
object Test2 extends App{
val conf = new SparkConf().setAppName("RVM").setMaster("local")
val sc = new SparkContext(conf)
val matrix = new DenseMatrix(2,2,Array(1.0,2,3,4))
new Test(sc,matrix).run()
}
class Test(scc:SparkContext,PHI:DenseMatrix) extends Serializable{
val ts = 0.1
def run(): Unit ={
val rdds = scc.parallelize(0 to 3)
val a = rdds.map(
x =>{
PHI.toArray.apply(x)*x
}
)
a.collect.foreach(println(_))
}
}
出现“org.apache.spark.SparkException: Task not serializable"这个错误，一般是因为在map、filter等的参数使用了外部的变量，但是这个变量不能序列化。特别是当引用了某个类（经常是当前类）的成员函数或变量时，会导致这个类的所有成员（整个类）都需要支持序列化。解决这个问题最常用的方法有：
- 如果可以，将依赖的变量放到map、filter等的参数内部定义。这样就可以使用不支持序列化的类；
- 如果可以，将依赖的变量独立放到一个小的class中，让这个class支持序列化；这样做可以减少网络传输量，提高效率；
- 如果可以，将被依赖的类中不能序列化的部分使用transient关键字修饰，告诉编译器它不需要序列化。
- 将引用的类做成可序列化的。
总结
通过上面填坑过程发现如下规律：
1、在rdd应该外部变量类实例的时候，类需要继承Serializable
2、在非静态类中（class声明的类），若是类属性是一个对象，则该属性不能在rdd里面直接使用，尽管该对象是已经继承了Serializable，可以直接在rdd前将该属性赋值为一个变量，再在rdd里面调用该变量