spark sql 列怎么转换数据类型

发布网友发布时间：2022-05-03 02:16

共2个回答

懂视网时间：2022-05-03 06:37

1.编写给ResultSet添加spark的schema成员及DF（DataFrame）成员

/*
 spark、sc对象因为是全局的，没有导入，需自行定义
 teradata的字段类型转换成spark的数据类型
*/

import java.sql.{ResultSet, ResultSetMetaData}

import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row}

object addDataframeMember {

 trait ResultSetMetaDataToSchema {
 def columnCount: Int

 def schema: StructType
 }

 implicit def wrapResultSetMetaData(rsmd: ResultSetMetaData) = {
 new ResultSetMetaDataToSchema {
 def columnCount = rsmd.getColumnCount

 def schema = {
 def tdCovert(tdDpeStr: String, precision: Int = 0, scale: Int = 0, className: String = ""): DataType = {
  tdDpeStr match {
  case "BYTEINT" => IntegerType
  case "SMALLINT" => ShortType
  case "INTEGER" => IntegerType
  case "BIGINT" => LongType
  case "FLOAT" => DoubleType
  case "CHAR" => CharType(precision)
  case "DECIMAL" => DecimalType(precision, scale)
  case "VARCHAR" => StringType
  case "BYTE" => ByteType
  case "VARBYTE" => ByteType
  case "DATE" => DateType
  case "TIME" => TimestampType
  case "TIMESTAMP" => TimestampType
  case "CLOB" => StringType
  case "BLOB" => BinaryType
  case "Structured UDT" => ObjectType(Class.forName(className))
  }
 }

 def col2StructField(rsmd: ResultSetMetaData, i: Int): StructField = StructField(rsmd.getColumnName(i), tdCovert(rsmd.getColumnTypeName(i), rsmd.getPrecision(i), rsmd.getScale(i), rsmd.getColumnClassName(i)), rsmd.isNullable(i) match { case 1 => true case 0 => false }).withComment(rsmd.getColumnLabel(i))

 def rsmd2Schema(rsmd: ResultSetMetaData): StructType = (1 to columnCount).map(col2StructField(rsmd, _)).foldLeft(new StructType)((s: StructType, i: StructField) => s.add(i))

 rsmd2Schema(rsmd)
 }
 }
 }

 trait ResultSetToDF {
 def schema: StructType

 def DF: DataFrame
 }

 implicit def wrapResultSet(rs: ResultSet) = {
 def rsmd = rs.getMetaData

 def toList[T](retrieve: ResultSet => T): List[T] = Iterator.continually((rs.next(), rs)).takeWhile(_._1).map(r => r._2).map(retrieve).toList

 def rsContent2Row(rs: ResultSet): Row = Row.fromSeq(Array.tabulate[Object](rsmd.columnCount)(i => rs.getObject(i + 1)).toSeq)

 new ResultSetToDF {
 def schema = rsmd.schema

 def DF = spark.createDataFrame(sc.parallelize(toList(rsContent2Row)), schema)
 }

 }


}

2.正常基于JDBC连接并且获得数据集游标

import java.sql.{Connection, DriverManager}

/*
 获取TeraData的连接
*/

val (dialect, host, user, passwd, database, charset) = ("teradata", "ip", "user", "password", "database", "ASCII")
val tdConf = collection.immutable.Map(
 "driver" -> "com.ncr.teradata.TeraDriver",
 "uri" -> s"jdbc:$dialect://$host/CLIENT_CHARSET=EUC_CN,TMODE=TERA,COLUMN_NAME=ON,CHARSET=ASCII,database=$database",
 "username" -> user,
 "password" -> passwd
)

def getTeraConn: Connection = {
 Class.forName(tdConf("driver"))
 DriverManager.getConnection(tdConf("uri"), tdConf("username"), tdConf("password"))
}
val sql = "SELECT TOP 10 * FROM xxx"
var conn = getTeraConn
val stmt = conn.createStatement()
val rs = stmt.executeQuery(sql)

3.导入隐式转换，调用成员

import addDataframeMember.wrapResultSet
rs.DF.show()

JDBC的ResultSet游标转spark的DataFrame，数据类型的映射以TeraData数据库为例

标签：har lse typename 编写 nec scala etc sci long

热心网友时间：2022-05-03 03:45

Java获取数据库的表中各字段的字段名，代码如下：
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.PreparedStatement;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
public class TestDemo {
public static Connection getConnection() {
Connection conn = null;
try {
Class.forName("com.mysql.jdbc.Driver");
String url = "jdbc:mysql://数据库IP地址:3306/数据库名称";
String user = "数据库用户名";
String pass = "数据库用户密码";
conn = DriverManager.getConnection(url, user, pass);
} catch (ClassNotFoundException e) {
e.printStackTrace();
} catch (SQLException e) {
e.printStackTrace();
}
return conn;
}
public static void main(String[] args) {
Connection conn = getConnection();
String sql = "select * from AccessType";
PreparedStatement stmt;
try {
stmt = conn.prepareStatement(sql);
ResultSet rs = stmt.executeQuery(sql);
ResultSetMetaData data = rs.getMetaData();
for (int i = 1; i <= data.getColumnCount(); i++) {
// 获得所有列的数目及实际列数
int columnCount = data.getColumnCount();
// 获得指定列的列名
String columnName = data.getColumnName(i);
// 获得指定列的列值
int columnType = data.getColumnType(i);
// 获得指定列的数据类型名
String columnTypeName = data.getColumnTypeName(i);
// 所在的Catalog名字
String catalogName = data.getCatalogName(i);
// 对应数据类型的类
String columnClassName = data.getColumnClassName(i);
// 在数据库中类型的最大字符个数
int columnDisplaySize = data.getColumnDisplaySize(i);
// 默认的列的标题
String columnLabel = data.getColumnLabel(i);
// 获得列的模式
String schemaName = data.getSchemaName(i);
// 某列类型的精确度(类型的长度)
int precision = data.getPrecision(i);
// 小数点后的位数
int scale = data.getScale(i);
// 获取某列对应的表名
String tableName = data.getTableName(i);
// 是否自动递增
boolean isAutoInctement = data.isAutoIncrement(i);
// 在数据库中是否为货币型
boolean isCurrency = data.isCurrency(i);
// 是否为空
int isNullable = data.isNullable(i);
// 是否为只读
boolean isReadOnly = data.isReadOnly(i);
// 能否出现在where中
boolean isSearchable = data.isSearchable(i);
System.out.println(columnCount);
System.out.println("获得列" + i + "的字段名称:" + columnName);
System.out.println("获得列" + i + "的类型,返回SqlType中的编号:"+ columnType);
System.out.println("获得列" + i + "的数据类型名:" + columnTypeName);
System.out.println("获得列" + i + "所在的Catalog名字:"+ catalogName);
System.out.println("获得列" + i + "对应数据类型的类:"+ columnClassName);
System.out.println("获得列" + i + "在数据库中类型的最大字符个数:"+ columnDisplaySize);
System.out.println("获得列" + i + "的默认的列的标题:" + columnLabel);
System.out.println("获得列" + i + "的模式:" + schemaName);
System.out.println("获得列" + i + "类型的精确度(类型的长度):" + precision);
System.out.println("获得列" + i + "小数点后的位数:" + scale);
System.out.println("获得列" + i + "对应的表名:" + tableName);
System.out.println("获得列" + i + "是否自动递增:" + isAutoInctement);
System.out.println("获得列" + i + "在数据库中是否为货币型:" + isCurrency);
System.out.println("获得列" + i + "是否为空:" + isNullable);
System.out.println("获得列" + i + "是否为只读:" + isReadOnly);
System.out.println("获得列" + i + "能否出现在where中:"+ isSearchable);
}
} catch (SQLException e) {
e.printStackTrace();
}
}
}