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

一.创建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()

执行效果如下：

备注：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()

执行结果：

从执行结果可知，直接使用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内部：

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)

执行结果：

根据执行结果可知，不管是在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。
执行结果如下：

2.把dataframe转换为rdd【按照1.x版本操作】

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

执行结果：

五.备注

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