Spark SQL读数据库时不支持某些数据类型的问题

public class Config {
  // spark-jdbc parameter names
  public static String JDBC_PARA_URL = "url";
  public static String JDBC_PARA_USER = "user";
  public static String JDBC_PARA_PASSWORD = "password";
  public static String JDBC_PARA_DRIVER = "driver";
  public static String JDBC_PARA_TABLE = "dbtable";
  public static String JDBC_PARA_FETCH_SIZE = "fetchsize";
}

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._

// 主类
object Main {

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder().master("yarn").appName("test")getOrCreate()
    val sqlContext = sparkSession.sqlContext
    val sc = sparkSession.sparkContext
    val partitionNum = 16
    val fetchSize = 1000
    val jdbcUrl = "..."
    val userName = "..."
    val schema_table = "..."
    val password = "..."
    val jdbcDriver = "oracle.jdbc.driver.OracleDriver"
    // 注意需要将oracle jdbc driver jar放置在spark lib jars目录下，或者spark2-submit提交spark application时添加--jars参数
    val jdbcDF = sqlContext.read.format("jdbc").options(
          Map(Config.JDBC_PARA_URL -> jdbcUrl,
            Config.JDBC_PARA_USER -> userName,
            Config.JDBC_PARA_TABLE -> schema_table,
            Config.JDBC_PARA_PASSWORD -> password,
            Config.JDBC_PARA_DRIVER -> jdbcDriver,
            Config.JDBC_PARA_FETCH_SIZE -> s"$fetchSize")).load()
    val rdd = jdbcDF.rdd
    rdd.count()
    ......
}

object SaicSparkJdbcDialect {


  def useMyJdbcDIalect(jdbcUrl:String,dbType:String): Unit ={

    val logger = LoggerFactory.getLogger(classOf[SaicSparkJdbcDialect])

    // 将当前的 JdbcDialect 对象unregistered掉
    val dialect = JdbcDialects
    JdbcDialects.unregisterDialect(dialect.get(jdbcUrl))

    if (dbType.equals("ORACLE")) {
      val OracleDialect = new JdbcDialect {
          // 只能处理ORACLE数据库
          override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle")
          // 修改数据库 SQLType 到 Spark DataType 的映射关系（从数据库读取到Spark中）
          override def getCatalystType(sqlType: Int, typeName: String, size: Int,
                                       md: MetadataBuilder): Option[DataType] = {
            if (sqlType==Types.TIMESTAMP || sqlType== -101 || sqlType== -102) {
              // 将不支持的 Timestamp with local Timezone 以TimestampType形式返回
              Some(TimestampType)
            } else if (sqlType == Types.BLOB) {
              Some(BinaryType)
            } else {
              Some(StringType)
            }
          }
          // 该方法定义的是数据库Spark DataType 到 SQLType 的映射关系，此处不需要做修改
          override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
            case StringType => Some(JdbcType("VARCHAR2(2000)", java.sql.Types.VARCHAR))
            case BooleanType => Some(JdbcType("NUMBER(1)", java.sql.Types.NUMERIC))
            case IntegerType => Some(JdbcType("NUMBER(10)", java.sql.Types.NUMERIC))
            case LongType => Some(JdbcType("NUMBER(19)", java.sql.Types.NUMERIC))
            case DoubleType => Some(JdbcType("NUMBER(19,4)", java.sql.Types.NUMERIC))
            case FloatType => Some(JdbcType("NUMBER(19,4)", java.sql.Types.NUMERIC))
            case ShortType => Some(JdbcType("NUMBER(5)", java.sql.Types.NUMERIC))
            case ByteType => Some(JdbcType("NUMBER(3)", java.sql.Types.NUMERIC))
            case BinaryType => Some(JdbcType("BLOB", java.sql.Types.BLOB))
            case TimestampType => Some(JdbcType("DATE", java.sql.Types.TIMESTAMP))
            case DateType => Some(JdbcType("DATE", java.sql.Types.DATE))
            case _ => None
          }
          override def quoteIdentifier(colName: String): String = {
            colName
          }
        }
        // register新创建的 JdbcDialect 对象
        JdbcDialects.registerDialect(OracleDialect)
      }