火花读取来自 SAS 国际移民组织的 JDBC

Question

我正在尝试使用火花 JDBC 从 SAS IOM 读取数据。问题是SAS JDBC驱动程序有点奇怪，所以我需要创建自己的方言：

object SasDialect extends JdbcDialect {
  override def canHandle(url: String): Boolean = url.startsWith("jdbc:sasiom")
  override def quoteIdentifier(colName: String): String = "\"" + colName + "\"n"
}

然而，这还不够。SAS区分了列标签（=人类可读的名称）和列名称（=您在SQL查询中使用的名称），但似乎Spark在模式发现中使用列标签而不是名称，请参阅下面的JdbcUtils摘录：

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala#L293

while (i < ncols) {
  val columnName = rsmd.getColumnLabel(i + 1)

这会导致SQL错误，因为它试图在生成的SQL代码中使用人类可读的列名。

为了使SAS IOM JDBC工作，这需要是getColumnName而不是getColumn标签。有没有办法用方言来说明这一点？除了包装整个com.sas.rio之外，我真的找不到一个方法来实现这一点。MVADriver和resultsetmeta

直率的

Answer 1

与此同时，我发现了如何做，所以只是发布以供参考。诀窍是注册你自己的方言，如下所示。

另外，SAS用空格填充所有的varchar列，所以我修剪所有的string列。

  def getSasTable(sparkSession: org.apache.spark.sql.SparkSession, tablename: String): org.apache.spark.sql.DataFrame = {                                       
    val host : String = "dwhid94.msnet.railb.be";                                                                                                               
    val port : String = "48593";                                                                                                                                
    val props = new java.util.Properties();                                                                                                                     
    props.put("user", CredentialsStore.getUsername("sas"))                                                                                                      
    props.put("password", CredentialsStore.getPassword("sas"))                                                                                                  
    props.setProperty("driver", "com.sas.rio.MVADriver")                                                                                                        
    val sasconurl : String =  String.format("jdbc:sasiom://%s:%s", host, port);                                                                                 
                                                                                                                                                                
    object SasDialect extends JdbcDialect {                                                                                                                     
      override def canHandle(url: String): Boolean = url.startsWith("jdbc:sasiom")                                                                              
      override def quoteIdentifier(colName: String): String = "\"" + colName + "\"n"                                                                            
    }                                                                                                                                                           
    JdbcDialects.registerDialect(SasDialect)                                                                                                                    
    val df = sparkSession.read                                                                                                                                  
      .option("url", sasconurl)                                                                                                                                 
      .option("driver", "com.sas.rio.MVADriver")                                                                                                                
      .option("dbtable", tablename)                                                                                                                             
      .option("user",CredentialsStore.getUsername("sas"))                                                                                                       
      .option("password",CredentialsStore.getPassword("sas"))                                                                                                   
      .option("fetchsize",100)                                                                                                                                  
      .format("jdbc")                                                                                                                                           
      .load()                                                                                                                                                   
                                                                                                                                                                
    val strippedDf = sparkSession.createDataFrame(df.rdd.map(r => Row(r.toSeq.map(x => x match {case s: String => s.trim; case _ => x}): _*)), df.schema);      
    return strippedDf;                                                                                                                                          
  }