Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】

本文涉及的产品
云数据库 RDS SQL Server,独享型 2核4GB
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】

RDD 转换得到 DataFrame

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

  1. 利用反射机制推断 RDD 模式
  2. 使用编程方式定义 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对象

image.png

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

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

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

  1. 制作表头 schema: StructType
  2. 制作表中记录 rowRDD: RDD[Row]
  3. 合并表头和记录 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()
  }

运行结果:


总结

       今天上午就学到这里,本想着今天专门看看StructType、StructField和Row这三个类的,没想到就在这节课。这一篇主要学了RDD对象向DataFrame对象的转换以及Spark SQL如何读取数据库、写入数据库。


       下午学完这一章最后的DataSet。

相关实践学习
基于CentOS快速搭建LAMP环境
本教程介绍如何搭建LAMP环境,其中LAMP分别代表Linux、Apache、MySQL和PHP。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
相关文章
|
3天前
|
SQL 关系型数据库 MySQL
【MySQL 数据库】7、SQL 优化
【MySQL 数据库】7、SQL 优化
23 0
|
5天前
|
SQL 关系型数据库 MySQL
数据库管理新星:Navicat、HeidiSQL和SQL Workbench的竞争格局
随着数据管理需求的不断增长,数据库管理工具的选择变得至关重要。在众多选择中,Navicat、HeidiSQL和SQL Workbench作为备受关注的数据库管理工具,各自展现出独特的特性和功能,竞争格局日益激烈。
17 0
|
15天前
|
SQL 存储 关系型数据库
sql数据库的相关概念与底层介绍
sql数据库的相关概念与底层介绍
32 0
|
4天前
|
SQL Oracle Java
怎样写出可在各种数据库间移植的SQL
国际标准没有对这些函数做规定,这种 SQL 语句就会和数据库相关而无法移植了,而这些函数在应用开发中还特别常见。 Hibernate 能将包括这些函数的标准 HQL 语句转换成不同数据库的 SQL,但 HQL 支持的函数太少,碰到不认识的函数还是要分别注册,这就丧失了可移植性;而且 HQL 能生成的 SQL 本身也比较简单,覆盖面太窄了。
|
1月前
|
SQL 人工智能 运维
数据库基础入门 — SQL排序与分页
数据库基础入门 — SQL排序与分页
17 0
|
1月前
|
SQL 人工智能 运维
数据库基础入门 — SQL运算符
数据库基础入门 — SQL运算符
15 0
|
1月前
|
SQL 人工智能 运维
数据库基础入门 — SQL
数据库基础入门 — SQL
27 0
|
SQL 消息中间件 分布式计算
通过Spark SQL实时归档SLS数据
我在前一篇文章介绍过基于Spark SQL实现对HDFS操作的实时监控报警。今天,我再举例说明一下如何使用Spark SQL进行流式应用的开发。
2482 0
|
1月前
|
SQL 算法 数据库
【数据库SQL server】关系数据库标准语言SQL之数据查询
【数据库SQL server】关系数据库标准语言SQL之数据查询
58 0
|
1月前
|
SQL 算法 数据库
【数据库SQL server】关系数据库标准语言SQL之视图
【数据库SQL server】关系数据库标准语言SQL之视图
57 0