spark sql read sqlserver、write carbon报错spark.sql.AnalysisException: unsupported type: BinaryType_sql server unsupported_type

还原报错前提User class threw exception: org.apache.spark.sql.AnalysisException: unsupported type: BinaryType;(jdbc读写)

SQL server表user
字段名 字段类型
Filed1 bigint
Filed2 char
Filed3 decimal
Filed4 datetime
Filed5 nvarchar(max)
Filed6 timestamp

Carbon表 person
字段名 字段类型
Filed1 Long
Filed2 string
Filed3 double
Filed4 string
Filed5 string
Filed6 string

tableName = “user”
//方法一：读
val jdbcDF: Dataset[Row] = session.read.format("jdbc")
  .option("url", CommonConfig.SQLSERVER_URL)
  .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
  .option("dbtable", tableName)
  .option("user", CommonConfig.SQLSERVER_USER)
  .option("password", CommonConfig.SQLSERVER_PASSWORD)
  .load()

jdbcDF.show(10)

writeTableName = “person”
//方法二：写
jdbcDF
.write
  .format("carbondata")
  .option("dbName", CommonConfig.CARBONDATA_DBTABLE)
  .option("tableName", writeTableName)
  .option("compress", "true")
  .mode(SaveMode.Overwrite)
  .save()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

上面demo适合spark sql 正常识别的字段值

此处读user表报错，原因是spark不识别SQL server的nvarchar(max)、timestamp字段类型

解决方法

当遇到timestamp、nvarchar(max)等不识别的二进制类型时
read时需要cast转换相应字段类型，并设置table别名

tableName = “(select Filed1, Filed2, Filed3, Filed4, cast(Filed5 as nvarchar(4000)) as Filed5, cast(Filed6 as nvarchar(4000)) as Filed6 from user) temp”

再调用读，可以发现show正常打印数据

写的时候需要转换下，写一个case对象类

case class Person (
                   Filed1: Long,
                   Filed2: String,
                   Filed3: Double,
                   Filed4: String,
                   Filed5: String,
                   Filed6: String
                 )

1
2
3
4
5
6
7
8
9

写方法改写具体如下

jdbcDF
  .as[Person]
  .write
  .format("carbondata")
  .option("dbName", CommonConfig.CARBONDATA_DBTABLE)
  .option("tableName", writeTableName)
  .option("compress", "true")
  .mode(SaveMode.Overwrite)
  .save()
1
2
3
4
5
6
7
8
9

完美解决问题

参考文章：
Sql server中的 nvarchar(max) 到底有多大？
针对SQL Server表的spark.read读取错误（通过JDBC连接）