Spark1.x升级Spark2.x常见异常【map】

绝地灬酷狼 2023-02-10 03:38 6阅读 0赞

## 一.创建Spark入口 ##

相较于Spark1.x，Spark2.x最明显的区别就是程序执行入口的区别了，从SparkContext变为SparkSession。相较于SparkContext，SparkSession对SparkContext进行了二次封装，把原有Spark1.x中的SQLContext和HiveContext进行了合并，默认为SQLContext，当需要访问Hive时，只需开启对Hive的支持即可【.enableHiveSupport()，当然hive的相关配置还是要设置的】，如下：

val spark = SparkSession
          .builder
          .appName(s"${this.getClass.getSimpleName}")
          .master("local[2]")
          .getOrCreate()

## 二.创建DataFrame数据集 ##

在Spark2.x中，获取数据集的方式和Spark1.x非常类似，只需从结构化数据源直接加载或从非结构化数据源转换即可，如下：

// 数据集直接的转换
        import spark.implicits._
    
        val technology = spark.sparkContext
          .textFile("D:\\software\\spark-2.4.4\\data\\sql\\dataframe.txt")
          .map(_.split(","))
          .map(row => Technology(row(0), row(1).toLong, row(2).toLong))
          .toDF()
    
        technology.show()

执行效果如下：  
![在这里插入图片描述][20200519193903446.png]  
备注：import spark.implicits.\_是为了可以直接使用toDF()算子，否则会直接报错。

## 三.常见异常之map ##

下面要说到重点的地方了，在Spark升级到2版本后，map异常是非常常见的，究其原因其实就是：在Spark1.x中使用map之类的算子会把dataframe转换为rdd，而在Spark2.x中会转换为dataset。相较于rdd，dataset最大的区别就在于Encoder了。

对于scala的使用者而言，scala代码中嵌入部分java数据类型是非常常见的，这部分java数据类型就是报错的主要来源！下面使用代码验证具体哪些情况会导致异常：  
1.直接使用map

technology.map(row => row.getAs[String]("name")).show()

执行结果：  
![在这里插入图片描述][2020051919500891.png]  
从执行结果可知，直接使用map是不会报错的，因此map不是报错的原因。  
2.使用Row

technology.map(row => Row(row.getAs[String]("name"))).show()

执行结果直接报错，如下：

Error:(34, 19) Unable to find encoder for type org.apache.spark.sql.Row. An implicit Encoder[org.apache.spark.sql.Row] is needed to store org.apache.spark.sql.Row instances in a Dataset. Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.
        technology.map(row => Row(row.getAs[String]("name"))).show()
                      ^

根据执行结果可知，报错原因是Row没有encoder导致的，即使Row本身就来自spark内部：  
![在这里插入图片描述][20200519195505306.png]  
3.使用toMap和在map内部使用java的Calendar

technology.map(row => { 
          val cal = Calendar.getInstance()
          val format = new SimpleDateFormat("yyyy-MM-dd")
          val date = format.format(cal.getTime).replace("-", "")
          (row.getAs[String]("name"), date)
        })
          .collect()
          .toMap[String, String]
          .foreach(println)

执行结果：  
![在这里插入图片描述][20200519201225240.png]  
根据执行结果可知，不管是在map内部使用java基本类型，还是在外部使用toMap都不会报错！  
因此，可以看出，报错不在乎使用的java还是scala数据类型，而在于是否可以encoder。

## 四.解决方案 ##

常用的解决方面有两种：  
1.对没有encoder的数据类型指定encoder。如下

implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Row]
        technology.map(row => Row(row.getAs[String]("name"))).show()

单独指定Row类型的encoder。  
执行结果如下：  
![在这里插入图片描述][20200519203226841.png]  
2.把dataframe转换为rdd【按照1.x版本操作】

technology.rdd.map(row => Row(row.getAs[String]("name"))).foreach(println)

执行结果：  
![在这里插入图片描述][20200519203556661.png]

## 五.备注 ##

不同人在代码中使用到的数据类型是多种多样的，要使用第一种方案，只需要把\[Row\]中的Row换成自己使用的类型即可；若使用第二种方案可以不用考虑这些，但这样会破坏Spark2.x的使用风格【相当于Spark2.x中穿插有Spark1.x】！

[20200519193903446.png]: https://img-blog.csdnimg.cn/20200519193903446.png
[2020051919500891.png]: https://img-blog.csdnimg.cn/2020051919500891.png
[20200519195505306.png]: https://img-blog.csdnimg.cn/20200519195505306.png
[20200519201225240.png]: https://img-blog.csdnimg.cn/20200519201225240.png
[20200519203226841.png]: https://img-blog.csdnimg.cn/20200519203226841.png
[20200519203556661.png]: https://img-blog.csdnimg.cn/20200519203556661.png