从 RDD 转换得到 DataFrame

Saprk 提供了两种方法来实现从 RDD 转换得到 DataFrame：

利用反射机制推断 RDD 模式
使用编程方式定义 RDD 模式

下面使用到的数据 people.txt ：

Tom, 21
Mike, 25
Andy, 18

1、利用反射机制推断 RDD 模式

在利用反射机制推断 RDD 模式的过程时，需要先定义一个 case 类，因为只有 case 类才能被 Spark 隐式地转换为DataFrame对象。

object Tese{
  
    // 反射机制推断必须使用 case 类,case class 必须放到main方法之外
    case class Person(name: String,age: Long)  //定义一个case类

def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("rdd to df 1")
      .getOrCreate()


    import spark.implicits._ //这里的spark不是org.apache.spark这个包 而是我们创建的SparkSession对象 它支持把一个RDD隐式地转换为一个 DataFrame对象


    val rdd: RDD[Person] = spark.sparkContext
      .textFile("data/sql/people.txt")
      .map(line => line.split(","))
      .map(t => Person(t(0), t(1).trim.toInt))

    // 将RDD对象转为DataFrame对象
    val df: DataFrame = rdd.toDF()

    df.createOrReplaceTempView("people")

    spark.sql("SELECT * FROM people WHERE age > 20").show()

    spark.stop()
  }
}

注意事项1：

case 类必须放到伴生对象下，main方法之外，因为在隐式转换的时候它会自动通过伴生对象名.case类名来调用case类，如果放到main下面就找不到了。

注意事项2：

import spark.implicits._

这里的spark不是org.apache.spark这个包 而是我们上面创建的SparkSession对象 它支持把一个RDD隐式地转换为一个 DataFrame对象

2、使用编程方式定义 RDD 模式

反射机制推断时需要定义 case class，但当无法定义 case 类时，就需要采用编程式来定义 RDD 模式了。这种方法看起来比较繁琐，但是很好用，不容易报错。

我们现在同样加载 people.txt 中的数据，生成 RDD 对象，再把RDD对象转为DataFrame对象，进行SparkSQL 查询。主要包括三个步骤：

制作表头 schema: StructType
制作表中记录 rowRDD: RDD[Row]
合并表头和记录 df:DataFramw

def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("rdd to df 2")
      .getOrCreate()

    //1.制作表头-也就是定义表的模式
    val schema: StructType = StructType(Array(StructField("name", StringType, true),
      StructField("age", IntegerType, true)))
    //2.加载表中的记录-也就是读取文件生成RDD
    val rowRdd: RDD[Row] = spark.sparkContext
      .textFile("data/sql/people.txt")
      .map(_.split(","))
      .map(attr => Row(attr(0), attr(1).trim.toInt))
    //3.把表头和记录拼接在一起
    val peopleDF: DataFrame = spark.createDataFrame(rowRdd, schema)

    peopleDF.createOrReplaceTempView("people")

    spark.sql("SELECT * FROM people WHERE age > 20").show()

    spark.stop()
  }

运行结果：

+----+---+
|name|age|
+----+---+
| Tom| 21|
|Mike| 25|
+----+---+

Spark SQL读取数据库

导入依赖

根据自己本地的MySQL版本导入对应的驱动。

注意：mysql8.0版本在JDBC中的url是：" com.mysql.cj.jdbc.Driver "

<dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.31</version>
        </dependency>

读取 MySQL 中的数据

def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("jdbc spark sql")
      .getOrCreate()

    val mysql: DataFrame = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .option("dbtable", "student")
      .option("user", "root")
      .option("password", "Yan1029.")
      .load()

    mysql.show()

    spark.stop()
  }

运行结果：

默认显示整张表

+---+----+---+---+
| id|name|age|sex|
+---+----+---+---+
|  1| Tom| 21| 男|
|  2|Andy| 20| 女|
+---+----+---+---+

向 MySQL 写入数据

def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("jdbc spark sql")
      .getOrCreate()

    //导入两条student信息
    val rdd: RDD[Array[String]] = spark.sparkContext
      .parallelize(Array("3 Mike 22 男", "4 Cindy 23 女"))
      .map(_.split(" "))

    //设置模式信息-创建表头
    val schema: StructType = StructType(Array(StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("sex", StringType, true)))

    //创建Row对象 每个 Row对象都是表中的一行-创建记录
    val rowRDD = rdd.map(stu => Row(stu(0).toInt, stu(1), stu(2).toInt, stu(3)))

    //创建DataFrame对象 拼接表头和记录
    val df = spark.createDataFrame(rowRDD, schema)

    //创建一个 prop 变量 用来保存 JDBC 连接参数
    val prop = new Properties()
    prop.put("user","root")
    prop.put("password","Yan1029.")
    prop.put("driver","com.mysql.cj.jdbc.Driver")

    //写入数据 采用 append 模式追加
    df.write.mode("append").jdbc("jdbc:mysql://localhost:3306/spark","spark.student",prop)

    spark.stop()
  }

运行结果：