我正在查询 mysql 表
val url = "jdbc:mysql://XXX-XX-XXX-XX-XX.compute-1.amazonaws.com:3306/pg_partner"
val driver = "com.mysql.jdbc.Driver"
val username = "XXX"
val password = "XXX"
var connection:Connection = DriverManager.getConnection(url, username, password)
val statement = connection.createStatement()
val patnerName = statement.executeQuery("SELECT id,name FROM partner")
我确实得到了结果patnerName
但我需要转换为数据框。
我可以通过以下代码打印数据:
while (patnerName.next) {
val id = patnerName.getString("id")
val name = patnerName.getString("name")
println("id = %s, name = %s".format(id,name))
}
现在我该如何转换patnerName
到数据框?
所以你必须分几个步骤来完成:
- 定义您的列并准备架构
val columns = Seq("id", "name")
val schema = StructType(List(
StructField("id", StringType, nullable = true),
StructField("name", StringType, nullable = true)
))
- 定义如何在每次迭代时将 ResultSet 中的每条记录转换为 Row
def parseResultSet(rs: ResultSet): Row = {
val resultSetRecord = columns.map(c => rs.getString(c))
Row(resultSetRecord:_*)
}
- 定义一个函数将 ResultSet 转换为 Iterator[Row]。它将使用您在上一步中定义的函数(当您在下一步中调用它时)。
def resultSetToIter(rs: ResultSet)(f: ResultSet => Row): Iterator[Row] =
new Iterator[Row] {
def hasNext: Boolean = rs.next()
def next(): Row = f(rs)
}
- 定义一个函数,该函数从 Iterator[Row].toSeq 创建 RDD,该函数使用您在上一步中定义的函数。使用 schema 从 RDD 创建 DataFrame
def parallelizeResultSet(rs: ResultSet, spark: SparkSession): DataFrame = {
val rdd = spark.sparkContext.parallelize(resultSetToIter(rs)(parseResultSet).toSeq)
spark.createDataFrame(rdd, schema) // use the schema you defined in step 1
}
- 最后调用你的函数
val df: DataFrame = parallelizeResultSet(patner, spark)
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)