大数据-94 Spark 集群 SQL DataFrame DataSet RDD 创建与相互转换 SparkSQL

news2024/11/14 12:06:37

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（正在更新！）

章节内容

上节完成的内容如下：

SparkSQL介绍
SparkSQL特点
SparkSQL数据抽象
SparkSQL数据类型

SparkSession

在 Spark2.0 之前

SQLContext 是创建 DataFrame 和执行SQL的入口
HiveContext 通过HiveSQL语句操作Hive数据，兼Hive操作，HiveContext继承自SQLContext

在 Spark2.0 后

这些入口点统一到了SparkSession，SparkSession封装了SQLContext及HiveContext
实现了SQLContext即HiveContext所有功能
通过SparkSession可以获取到SparkContext

RDD（Resilient Distributed Dataset，弹性分布式数据集）

RDD 是 Spark 的基础抽象，它表示一个不可变的、分布式的数据集。

特点：

不可变性：RDD 是不可变的，一旦创建就不能修改。任何对 RDD 的操作都会生成一个新的 RDD。
弹性：RDD 可以自动从节点失败中恢复数据，通过将计算逻辑重新应用到原始数据来重建丢失的数据。
分布式：RDD 可以分布在多个节点上执行操作，充分利用集群的计算能力。
延迟计算：RDD 的操作是延迟执行的（lazy evaluation），即只有在触发行动操作（如 count()、collect()）时，Spark 才会实际执行计算。
类型安全：RDD 是类型化的，但它的 API 是松散类型（loosely typed）的，这意味着编译器不会在编译时检查数据的类型，而是在运行时才会发现类型错误。

DataFrame

DataFrame 是一种基于 RDD 的分布式数据集，它具有命名的列。

特点：

结构化数据：DataFrame 是一个二维表格，具有命名的列和行，类似于关系数据库中的表或 Pandas 的 DataFrame。
优化引擎：DataFrame 受益于 Spark SQL 引擎的优化，如 Catalyst 优化器，可以自动优化查询并生成高效的执行计划。
丰富的 API：DataFrame 提供了一个高层次的 API，支持复杂的查询、过滤、聚合和连接操作。
类型不安全：与 RDD 不同，DataFrame 是动态类型（dynamic typing）的，数据类型检查是在运行时进行的，因此它在编译时不进行类型检查。

DataSet

DataSet 是 Spark 1.6 引入的一个新的数据抽象，它结合了 RDD 的强类型优势和 DataFrame 的优化能力。

特点：

类型安全：DataSet 是强类型的，它利用编译时类型检查，确保在编译时检测类型错误。
优化和性能：DataSet 受益于 Catalyst 优化器和 Tungsten 执行引擎，提供与 DataFrame 相同的优化能力，同时保留了类型安全性。
更丰富的 API：DataSet 提供了 RDD 的大部分 API，如 map、filter 等，同时也支持 SQL 查询。
统一 API：DataSet API 统一了 RDD 和 DataFrame，提供了一种更具表现力和安全性的编程模型。

DataFrame & Dataset 创建

不要刻意区分： DF & DS，DF是一种特殊的DS：ds.transformation => ds

由 Range 生成 Dataset

在 spark-shell 中进行测试

val numDS = spark.range(5, 100, 5)
// orderBy 转换操作 
numDS.orderBy(desc("id")).show(5)
// 统计信息
numDS.describe().show
// 显示 Schema 信息
numDS.printSchema
// 使用RDD执行同样的操作
numDS.rdd.map(_.toInt).stats
// 检查分区数
numDS.rdd.getNumPartitions

运行测试的过程如下图所示：
在这里插入图片描述

有集合生成Dataset

Dataset = RDD[case class]，在 spark-shell 中进行测试

case class Person(name: String, age: Int, height: Int)

// 注意 Seq 中元素的类型
val seq1 = Seq(Person("Jack", 28, 184), Person("Tom", 10, 144), Person("Andy", 16, 165))

val ds1 = spark.createDataset(seq1)
ds1.printSchema
ds1.show

执行的结果：
在这里插入图片描述
再来一个测试：

val seq2 = Seq(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))
val ds2 = spark.createDataset(seq2)
ds2.printSchema
ds2.show

执行的结果：
在这里插入图片描述

由集合生成DataFrame

DataFrame = RDD[Row] + Schema
继续进行测试：

val lst = List(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))
val df1 = spark.createDataFrame(lst).withColumnRenamed("_1", "name1").withColumnRenamed("_2", "age1").withColumnRenamed("_3", "height1")
df1.orderBy("age1").show(10)

执行的结果如下图所示：
在这里插入图片描述

RDD转成DataFrame

DataFrame = RDD[Row] + Schema

val arr = Array(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))
val rdd1 = sc.makeRDD(arr).map(f => Row(f._1, f._2, f._3))

val schema = StructType(
  StructField("name", StringType, false) ::
  StructField("age", IntegerType, false) ::
  StructField("height", IntegerType, false) ::
  Nil
)

val schema1 = (new StructType).add("name", "string", false).add("age", "int", false).add("height", "int", false)
val rddToDF = spark.createDataFrame(rdd1, schema)
rddToDF.orderBy(desc("name")).show(false)

执行的结果如下图：
在这里插入图片描述

RDD转Dataset

Dataset = RDD[case class]
DataFrame = RDD[Row] + Schema

val arr = Array(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))
val rdd1 = sc.makeRDD(arr)
val ds2 = spark.createDataset(rdd1)
ds2.show(10)

执行的结果如下图：
在这里插入图片描述

从文件创建DataFrame

CSV文件

我们生成了一个CSV文件，大致内容如下：
在这里插入图片描述

运行测试

val df1 = spark.read.csv("/opt/wzk/data/people1.csv")
df1.printSchema()
df1.show()

运行结果如下图所示：
在这里插入图片描述

三者转换

在这里插入图片描述
Spark SQL 提供了一个领域特定语言（DSL）以方便操作结构化数据，核心思想还是SQL，仅仅是一个语法问题。

RDD 与 DataFrame 之间的转换

RDD 转换为 DataFrame

将 RDD 转换为 DataFrame 需要提供数据的模式信息。通常你会使用 toDF() 方法将 RDD 转换为 DataFrame。
这里有两种主要方法：

使用隐式转换：需要导入 spark.implicits._，这允许你在不显式提供模式的情况下将常见的 RDD（如元组）转换为 DataFrame。
使用 StructType 定义模式：如果 RDD 的数据结构比较复杂，或者你需要精确控制 DataFrame 的模式，可以使用 StructType 和 Row。

DataFrame 转换为 RDD：