spark3.0教程-SparkSQL：RDD转DataFrame

作者：马育民 • • 阅读：10139

说明

在开发中，可能会遇到复杂情况，需要先通过 RDD 对数据进行处理，然后使用 SparkSQL 处理起来才 更方便、容易

此时就需要将 RDD 转成 DataFrame

下面代码上接：spark3.0-SparkSQL 第一个idea工程

方式一

通过 toDF() 方法，推断出rdd的schema，默认是string类型

缺点：由于类型都是 string，虽然简单，但不建议使用

例子

package rdd2dataframe

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object Rdd2DataFrame {
  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()

    // 读取文本文件，获取一行一行的数据
    // 泛型必须是元组，不能是Array、List
    val rdd: RDD[(String,String,String,String,String)] = sc.textFile("cdz.txt")
      .map( line => {
        val arr: Array[String] = line.split(",")
        (arr(0),arr(1),arr(2),arr(3),arr(4))
      })

    // 导入隐饰操作，否则RDD无法调用toDF方法。导入 spark对象的隐式操作
    import spark.implicits._
    //推断rdd的schema，默认是string类型，简单但不建议使用
    val df: DataFrame = rdd.toDF("id","ele_num","time","start_time","end_time")


    df.printSchema()
    df.show()
    spark.close()
  }
}

执行结果：

root
 |-- id: string (nullable = true)
 |-- ele_num: integer (nullable = true)
 |-- time: integer (nullable = true)
 |-- start_time: timestamp (nullable = true)
 |-- end_time: timestamp (nullable = true)

+---+-------+----+-------------------+-------------------+
| id|ele_num|time|         start_time|           end_time|
+---+-------+----+-------------------+-------------------+
|  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|
+---+-------+----+-------------------+-------------------+

方式二

通过 SparkSession 的 createDataFrame() 方法创建

方法一

createDataFrame 声明

通过 StructType 定义列类型 ，创建 DataFrame 对象

def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame

创建 StructType

//通过StructType定义列类型，创建schema信息
val structSchema: StructType = StructType(
  List(
    StructField("id", StringType, true),
    StructField("ele_num", IntegerType, true),
    StructField("time", IntegerType, true),
    StructField("start_time", TimestampType, true),
    StructField("end_time", TimestampType, true)
  )
)

例子

package rdd2dataframe

import java.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object Rdd2DataFrame2 {
  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")

    //通过StructType定义列类型
    //创建schema信息
    val structSchema: StructType = StructType(
      List(
        StructField("id", StringType, true),
        StructField("ele_num", IntegerType, true),
        StructField("time", IntegerType, true),
        StructField("start_time", TimestampType, true),
        StructField("end_time", TimestampType, true)
      )
    )
    //注意将数据格式化为 structSchema 相应的类型
    val rowRDD: RDD[Row] = file.map(line=> {
      val arr: Array[String] = line.split(",")
      val startDate: sql.Timestamp = java.sql.Timestamp.valueOf(arr(3).trim)
      val endDate: sql.Timestamp = java.sql.Timestamp.valueOf(arr(4).trim)
      val row = Row(arr(0).trim, arr(1).toInt, arr(2).trim.toInt, startDate, endDate)
      row

    })
    //创建dataframe
    val df: DataFrame = spark.createDataFrame(rowRDD,structSchema)

    df.printSchema()
    df.show()
    spark.close()
  }
}

方法二

通过定义 case class 样例类 实现

createDataFrame 声明

def createDataFrame[A <: Product : TypeTag](rdd: RDD[A]): DataFrame

定义样例类

//定义样例类，封装数据，这里用 Integer，可以是空，不要用 Int
case class Cdz(id: String, eleNum: Integer, time:Integer, startTime: Timestamp, endTime: Timestamp) {

}

主类

package rdd2dataframe

import java.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

object Rdd2DataFrame3 {
  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")

    //通过定义类格式化为dataframe
    val rowRDD: RDD[Cdz] = file.map(line=> {
      val arr: Array[String] = line.split(",")
      val startDate: sql.Timestamp = java.sql.Timestamp.valueOf(arr(3).trim)
      val endDate: sql.Timestamp = java.sql.Timestamp.valueOf(arr(4).trim)
      val row = Cdz(arr(0).trim, arr(1).toInt, arr(2).trim.toInt, startDate, endDate)
      row

    })
    //创建dataframe
    val df: DataFrame = spark.createDataFrame(rowRDD)

    df.printSchema()
    df.show()
    spark.close()
  }
}

原文出处：/show_1IX4K2m4KQv8.html

吉ICP备15003043号