说明
在开发中，可能会遇到复杂情况，需要先通过 RDD 对数据进行处理，然后使用 SparkSQL 处理起来才 更方便、容易
此时就需要将 RDD 转成 Dataset
下面代码上接：spark3.0-SparkSQL 第一个idea工程
方式一
通过 toDS() 方法
例子：元组
package rdd2dataframe
import java.sql.Timestamp
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
object Rdd2Dataset {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
val sc = new SparkContext(sparConf)
//创建SparkSession
val spark: SparkSession = SparkSession.builder().config(sparConf).getOrCreate()
// 读取文本文件，获取一行一行的数据
val file: RDD[String] = sc.textFile("cdz.txt")
val rdd: RDD[(String, Integer, Integer, Timestamp, Timestamp)] = file.map(line => {
val arr: Array[String] = line.split(",")
val startDate: Timestamp = java.sql.Timestamp.valueOf(arr(3).trim)
val endDate: Timestamp = java.sql.Timestamp.valueOf(arr(4).trim)
(arr(0).trim, arr(1).toInt, arr(2).trim.toInt, startDate, endDate)
})
import spark.implicits._
// 自动推断
val ds: Dataset[(String, Integer, Integer, Timestamp, Timestamp)] = rdd.toDS()
ds.printSchema()
ds.show()
spark.close()
}
}
执行结果：
root
|-- id: string (nullable = true)
|-- eleNum: integer (nullable = false)
|-- Time: integer (nullable = false)
|-- startTime: timestamp (nullable = true)
|-- endTime: timestamp (nullable = true)
+---+------+----+-------------------+-------------------+
| id|eleNum|Time| startTime| endTime|
+---+------+----+-------------------+-------------------+
| 1| 100| 10|2020-01-01 09:30:50|2020-01-01 09:40:50|
| 2| 200| 25|2020-01-11 10:40:50|2020-01-11 11:05:50|
| 3| 150| 12|2020-01-01 11:20:50|2020-01-01 11:32:50|
| 1| 300| 50|2020-01-03 14:30:50|2020-01-03 15:20:10|
| 1| 50| 5|2020-02-02 12:30:50|2020-02-02 12:35:50|
| 2| 500| 50|2020-02-02 15:30:30|2020-02-02 17:20:30|
| 3| 150| 12|2020-02-01 11:20:50|2020-02-01 11:32:50|
| 3| 150| 12|2020-02-03 11:20:50|2020-02-03 11:32:50|
| 1| 100| 10|2020-03-01 09:30:50|2020-03-01 09:40:50|
| 1| 100| 10|2020-03-05 09:30:50|2020-03-05 09:40:50|
| 2| 500| 120|2020-03-12 15:30:30|2020-03-12 17:30:30|
| 2| 500| 120|2020-03-22 15:30:30|2020-03-22 17:30:30|
| 3| 400| 50|2020-03-02 17:30:20|2020-03-02 18:20:20|
| 4| 100| 5|2020-04-01 11:20:50|2020-04-01 11:25:50|
| 4| 50| 8|2020-04-21 11:20:50|2020-04-21 11:28:50|
| 5| 150| 12|2020-04-11 11:20:50|2020-04-11 11:32:50|
| 6| 150| 20|2020-04-15 11:20:50|2020-04-15 11:40:50|
+---+------+----+-------------------+-------------------+
例子：样例类
定义样例类：
package rdd2dataframe
import java.sql.Timestamp
//定义样例类，封装数据
case class Cdz(id: String, eleNum: Integer, time:Integer, startTime: Timestamp, endTime: Timestamp) {
}
主类：
package rdd2dataframe
import java.sql.Timestamp
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
object Rdd2Dataset3 {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
val sc = new SparkContext(sparConf)
//创建SparkSession
val spark: SparkSession = SparkSession.builder().config(sparConf).getOrCreate()
// 读取文本文件，获取一行一行的数据
val file: RDD[String] = sc.textFile("cdz.txt")
val rdd: RDD[Cdz] = file.map(line => {
val arr: Array[String] = line.split(",")
val startDate: Timestamp = java.sql.Timestamp.valueOf(arr(3).trim)
val endDate: Timestamp = java.sql.Timestamp.valueOf(arr(4).trim)
Cdz(arr(0).trim, arr(1).toInt, arr(2).trim.toInt, startDate, endDate)
})
import spark.implicits._
// 自动推断
val ds: Dataset[Cdz] = rdd.toDS()
ds.printSchema()
ds.show()
spark.close()
}
}
方式二
通过定义 SparkSession 的 createDataset() 方法创建
案例
package rdd2dataframe
import java.sql.Timestamp
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
object Rdd2Dataset2 {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
val sc = new SparkContext(sparConf)
//创建SparkSession
val ss: SparkSession = SparkSession.builder().config(sparConf).getOrCreate()
// 读取文本文件，获取一行一行的数据
val file: RDD[String] = sc.textFile("cdz.txt")
val rdd: RDD[(String, Integer, Integer, Timestamp, Timestamp)] = file.map(line => {
val arr: Array[String] = line.split(",")
val startDate: Timestamp = java.sql.Timestamp.valueOf(arr(3).trim)
val endDate: Timestamp = java.sql.Timestamp.valueOf(arr(4).trim)
(arr(0).trim, arr(1).toInt, arr(2).trim.toInt, startDate, endDate)
})
import spark.implicits._
//创建dataframe
val ds: Dataset[(String, Integer, Integer, Timestamp, Timestamp)] = ss.createDataset(rdd)
ds.printSchema()
ds.show()
ss.close()
}
}