Spark中使用take截取数据时报错：java.io.NotSerializableException

矫情吗；* 2021-11-11 12:12 435阅读 0赞

# 发生场景 #

在使用Spark读取HBase中的数据时  
不使用take函数截取数据时  
读取到的HBase中的数据可以正常输出  
但是使用take操作截取数据时  
就会抛出异常  
java.io.NotSerializableException: org.apache.hadoop.hbase.io.ImmutableBytesWritable  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gxMDI1MzcyNjQ1_size_16_color_FFFFFF_t_70]

# 发生原因 #

Spark分为两大角色Driver与Executor  
不使用take时，输出直接在Executor上，不需要走网络，输出时进行了序列化  
使用take后，将取到的数据返回给Driver，需要走网络，走了网络之后，要进行序列化，否则就会出错

# 解决办法 #

在获取spark配置对象时，对需要进行序列化的类进行注册

var sparkConf: SparkConf = new SparkConf().setAppName("SparkAppTemplate").
          setMaster("local[2]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          //告知那些类型需要序列化
          .registerKryoClasses(Array(classOf[ImmutableBytesWritable], classOf[Result]))

# 完整代码 #

import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.hbase.{CellUtil, HBaseConfiguration}
    import org.apache.hadoop.hbase.io.ImmutableBytesWritable
    import org.apache.hadoop.hbase.mapreduce.TableInputFormat
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.hadoop.hbase.client.Result
    import org.apache.hadoop.hbase.util.Bytes
    import org.apache.spark.rdd.RDD
    
    object ReadHBaseTableDataSpark {
      def main(args: Array[String]): Unit = {
    
        var sparkConf: SparkConf = new SparkConf().setAppName("SparkAppTemplate").
          setMaster("local[2]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          //告知那些类型需要序列化
          .registerKryoClasses(Array(classOf[ImmutableBytesWritable], classOf[Result]))
    
        //创建sparkContext对象：主要用于读取需要处理的数据，封装在RDD集合中；调度jobs执行
        val sc = new SparkContext(sparkConf)
        sc.setLogLevel("WARN")
    
        //第一步：数据的读取（输入）
        val conf: Configuration = HBaseConfiguration.create
        //设置读取的表
        conf.set(TableInputFormat.INPUT_TABLE, "ns1:orders")
        val resultRDD: RDD[(ImmutableBytesWritable, Result)] =sc.newAPIHadoopRDD(
          conf,
          classOf[TableInputFormat],         //设置数据读入格式
          classOf[ImmutableBytesWritable],   //设置输入数据key的类型
          classOf[Result]                    //设置输入数据value的类型
        )
        //第二步：数据的处理（分析）
        println(resultRDD.count())
    
        //取出数据
        resultRDD.take(3).foreach{case(key,result)=>{
          val rowkey = Bytes.toString(key.get())
          //输出rowkey
          println(rowkey)
          //获取每一个字段的值
          for(cell<-result.rawCells){
            val cf= Bytes.toString(CellUtil.cloneFamily(cell))
            val filed = Bytes.toString(CellUtil.cloneQualifier(cell))
            val value =Bytes.toString(CellUtil.cloneValue(cell))
            println(cf+":"+filed+":"+value)
          }
        }}
        //第三步：数据的输出（输出）
    
        //开发测试的时候，为了对每个spark app页面监控查看job的执行情况，
        //spark app运行结束4040端口就没了
    
        //关闭资源
        sc.stop()
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gxMDI1MzcyNjQ1_size_16_color_FFFFFF_t_70]: /images/20211109/25de3c1927a54d7c975a7d296c13e6ba.png