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

2019 年 12 月 5 日
筆記

在大数据平台中，经常需要做数据的ETL，从传统关系型数据库RDBMS中抽取数据到HDFS中。之前开发数据湖新版本时使用Spark SQL来完成ETL的工作，但是遇到了 Spark SQL 不支持某些数据类型（比如ORACLE中的Timestamp with local Timezone）的问题。

一、系统环境

Spark 版本：2.1.0.cloudera1
JDK 版本：Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_131
ORACLE JDBC driver 版本：ojdbc7.jar
Scala 版本：2.11.8

二、Spark SQL读数据库表遇到的不支持某些数据类型

Spark SQL 读取传统的关系型数据库同样需要用到 JDBC，毕竟这是提供的访问数据库官方 API。Spark要读取数据库需要解决两个问题：

分布式读取；
原始表数据到DataFrame的映射。

2.1 业务代码

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()      ......  }

2.2 部分数据类型不支持

比如ORACLE中的Timestamp with local Timezone 和 FLOAT(126)。

三、解决方法：自定义JdbcDialects

3.1 什么是JdbcDialects ？

Spark SQL 中的 org.apache.spark.sql.jdbc package 中有个类 JdbcDialects.scala，该类定义了Spark DataType 和 SQLType 之间的映射关系，分析该类的源码可知，该类是一个抽象类，包含以下几个方法：

def canHandle(url : String)：判断该JdbcDialect 实例是否能够处理该jdbc url；
getCatalystType(sqlType: Int, typeName: String, size: Int, md: MetadataBuilder)：输入数据库中的SQLType，得到对应的Spark DataType的mapping关系；
getJDBCType(dt: DataType)：输入Spark 的DataType，得到对应的数据库的SQLType；
quoteIdentifier(colName: String)：引用标识符，用来放置某些字段名用了数据库的保留字（有些用户会使用数据库的保留字作为列名）；
其他……。

该类还有一个伴生对象，其中包含3个方法：

get(url: String)：根据database的url获取JdbcDialect 对象；
unregisterDialect(dialect: JdbcDialect)：将已注册的JdbcDialect 注销；
registerDialect(dialect: JdbcDialect)：注册一个JdbcDialect。

3.2 解决步骤

使用get(url: String)方法获取当前的 JdbcDialect 对象；
将当前的 JdbcDialect 对象 unregistered 掉；
new 一个 JdbcDialect 对象，并重写方法（主要是getCatalystType()方法，因为其定义了数据库 SQLType 到 Spark DataType 的映射关系），修改映射关系，将不支持的 SQLType 以其他的支持的数据类型返回比如StringType，这样就能够解决问题了；
register新创建的 JdbcDialect 对象。

3.3 解决方案的业务代码

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)        }

本文来自：https://www.jianshu.com/p/20b82891aac9