说明
只是读取csv文件，默认是字段是 string 类型，只能做简单查询，可以通过 样例类 指定列类型
关键代码
val df: DataFrame = spark.read
.option("header", "false") //默认为false，表示第一行不是表头
.option("sep", ",") //列分隔符
.schema(ScalaReflection.schemaFor[样例类名].dataType.asInstanceOf[StructType]) // 指定类型
.csv("cdz.txt") //读取文件
根据 样例类的 主构造器形参的顺序 匹配 文件的 列顺序
案例
定义样例类
//定义样例类，封装数据
case class Cdz(id: String, eleNum: Integer, time:Integer, startTime: Timestamp, endTime: Timestamp)
注意：样例类的 主构造器形参的顺序 要与 文件的 列顺序 一致
主类
package rdd2dataframe
import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
object Std2_pojo {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
//创建SparkSession
val spark: SparkSession = SparkSession.builder().config(sparConf).getOrCreate()
// 读取文本文件，获取一行一行的数据
val df: DataFrame = spark.read
.option("header", "false") //默认为false，表示第一行不是表头
.option("sep", ",") //列分隔符
.schema(ScalaReflection.schemaFor[Cdz].dataType.asInstanceOf[StructType]) // 指定类型
.csv("cdz.txt") //读取文件
df.printSchema()
df.show()
spark.close()
}
}
执行结果
Schema：
root
|-- id: string (nullable = true)
|-- eleNum: integer (nullable = true)
|-- Time: integer (nullable = true)
|-- 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|
+---+------+----+-------------------+-------------------+
查询
接着上面：
// 创建临时视图
df.createOrReplaceTempView("cdz")
val res: DataFrame = spark.sql("select id,sum(time) as time_sum from cdz group by id")
res.show()