百度360必应搜狗淘宝本站头条
mysql distinctjenkins官网mysql 子查询vba splitpython list sort
当前位置：网站首页 > IT技术 > 正文

SparkSQL——DataFrame的创建与使用

wptr33 2025-04-06 22:17 82 浏览

本文主要从以下几个方面介绍SparkSQL中的DataFrame:

第一，SparkSQL的作用

第二，什么是DataFrame

第三，DataFrame与RDD的区别

第四，DataFrame的创建与使用 （Spark1.x与Spark2.x两种不同版本）


第一，SparkSQL的作用

SparkSQL是spark处理结构化数据的一个模块，它的前身是shark，与基础的spark RDD不同，SparkSQL提供了结构化数据及计算结果等信息的接口，在内部，SparkSQL使用这个额外的信息去执行额外的优化，有几种方式可以跟SparkSQL进行交互，包括SQL和DataSet API，使用相同的执行引擎进行计算的时候，无论是使用哪一种计算引擎都可以快速的计算。

在使用RDD进行处理时，需要了解RDD的每个算子的特点，以求得高效的执行相应的操作。在使用SparkSQL时，程序会自动优化算子的执行过程以及使用哪个算子，提高效率。SparkSQL主要用于进行结构化数据的处理，作为分布式的SQL查询引擎。


第二，什么是DataFrame

DataFrame的前身是SchemaRDD，从Spark 1.3.0开始SchemaRDD更名为DataFrame。与SchemaRDD的主要区别是：DataFrame不再直接继承自RDD，而是自己实现了RDD的绝大多数功能。你仍旧可以在DataFrame上调用rdd方法将其转换为一个RDD。DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库的二维表格，DataFrame带有Schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。

1.DataFrame的作用：

是Spark SQL提供的最核心的编程抽象。

以列的形式组织的，分布式的数据集合。

它其实和关系型数据库中的表非常类似，但是底层做了很多的优化。

2.DataFrame可以构建的来源：

（1）结构化的数据文件

（2）Hive中的表

（3）外部的关系型数据库

（4）RDD

第三，DataFrame与RDD的区别

RDD可看作是分布式的对象的集合，Spark并不知道对象的详细模式信息，DataFrame可看作是分布式的Row对象的集合，其提供了由列组成的详细模式信息，使得Spark SQL可以进行某些形式的执行优化。DataFrame和普通的RDD的逻辑框架区别如下所示：


左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解 Person类的内部结构。

而右侧的DataFrame却提供了详细的结构信息，使得Spark SQL可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。DataFrame多了数据的结构信息，即schema。这样看起来就像一张表了，DataFrame还配套了新的操作数据的方法，DataFrame API（如df.select())和SQL(select id, name from xx_table where ...)。

此外DataFrame还引入了off-heap,意味着JVM堆以外的内存, 这些内存直接受操作系统管理（而不是JVM）。Spark能够以二进制的形式序列化数据(不包括结构)到off-heap中, 当要操作数据时, 就直接操作off-heap内存. 由于Spark理解schema, 所以知道该如何操作。

RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化。

有了DataFrame这个高一层的抽象后，我们处理数据更加简单了，甚至可以用SQL来处理数据了，对开发者来说，易用性有了很大的提升。

不仅如此，通过DataFrame API或SQL处理数据，会自动经过Spark 优化器（Catalyst）的优化，即使你写的程序或SQL不高效，也可以运行的很快。

1、DataFrame与RDD的优缺点

RDD的优缺点：

优点:

（1）编译时类型安全

编译时就能检查出类型错误

（2）面向对象的编程风格

直接通过对象调用方法的形式来操作数据

缺点:

（1）序列化和反序列化的性能开销

无论是集群间的通信, 还是IO操作都需要对对象的结构和数据进行序列化和反序列化。

关于序列化问题可以参考这篇文章：
https://www.toutiao.com/i6846761221337809419/

（2）GC的性能开销

频繁的创建和销毁对象, 势必会增加GC

DataFrame通过引入schema和off-heap（不在堆里面的内存，指的是除了不在堆的内存，使用操作系统上的内存），解决了RDD的缺点, Spark通过schame就能够读懂数据, 因此在通信和IO时就只需要序列化和反序列化数据, 而结构的部分就可以省略了；通过off-heap引入，可以快速的操作数据，避免大量的GC。但是却丢了RDD的优点，DataFrame不是类型安全的, API也不是面向对象风格的。

第四，DataFrame的创建与使用

spark1.x版本

在该版本下，我选择使用的scala2.10.7版本

spark1.x版本中总共有三种编程模式：

1、RDD数据类型调用toDF方法将RDD转化为DataFrame

package xxx

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

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联case class，将非结构化数据转化为结构化数据
 * 3、显示的调用toDF方法将RDD转化为DataFrame
 * 4、注册临时表
 * 5、执行SQL（Transformation方式）
 * 6、执行action
 */
object SqlDemo1 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val boyRDD: RDD[Person] = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Person(id, name, age, score)
    })

    //该RDD装的是Person类型的数据，有了shcma信息，但是还是一个RDD
    //将RDD转换成DataFrame
    //导入隐式转换
    import sqlContext.implicits._
    val bdf: DataFrame = boyRDD.toDF

    //变成DF后就可以使用两种API进行编程了
    //把DataFrame先注册临时表
    bdf.registerTempTable("t_boy")

    //书写SQL（SQL方法应其实是Transformation）
    val result: DataFrame = sqlContext.sql("SELECT * FROM t_boy ORDER BY score desc, age asc")

    //查看结果（触发Action）
    result.show()

    sc.stop()

  }

  case class Person(id: Int, name: String, age: Int, score: Int)
}

2、在RDD中关联Row，将非结构化的数据转化为结构化的数据，然后定义模式，通过调用SqlContext的createDataFrame方法将RDD转化为DataFrame

package xxx

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StructField, StructType, StringType}
import org.apache.spark.sql.{DataFrame, Row, SQLContext, types}

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用sqlContext的createDataFrame方法
 * 5、注册临时表
 * 6、执行SQL（Transformation方式）
 * 7、执行action
 */
object SqlDemo2 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Row(id, name, age, score)
    })

    // 设置结构类型，表头信息
    val structType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", IntegerType, true)
    ))

    // 将RowRDD关联schema
    val bdf = sqlContext.createDataFrame(rowRDD, structType)

    //变成DF后就可以使用两种API进行编程了
    //把DataFrame先注册临时表
    bdf.registerTempTable("t_boy")

    //书写SQL（SQL方法应其实是Transformation）
    val result: DataFrame = sqlContext.sql("SELECT * FROM t_boy ORDER BY score desc, age asc")

    //查看结果（触发Action）
    result.show()

    sc.stop()

  }

}

3、创建DataFrame的方式同1，只是在使用DataFrame的时候不使用SQL语句，而是使用DataFrame API，这样就不需要将DataFrame注册成临时表

package xxx

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}

/**
 * Spark1.x 编程模式
 * 1、创建SparkContext,然后再创建SQLContext
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用sqlContext的createDataFrame方法
 * 6、执行DataFrame语句（Transformation方式）
 * 7、执行action
 */
object SqlDemo3 {
  def main(args: Array[String]): Unit = {
    //提交的这个程序可以连接到Spark集群中
    val conf = new SparkConf().setAppName("SQLDemo1").setMaster("local[2]")

    //创建SparkSQL的连接（程序执行的入口）
    // sparkContext不能创建特殊的RDD（DataFrame）
    val sc = new SparkContext(conf)

    //将SparkContext包装进而增强
    // 创建特殊的RDD（DataFrame），就是有schema信息的RDD
    val sqlContext = new SQLContext(sc)

    //先有一个普通的RDD，然后在关联上schema，进而转成DataFrame
    val lines = sc.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toInt
      Row(id, name, age, score)
    })

    // 设置结构类型，表头信息
    val structType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", IntegerType, true)
    ))

    // 将RowRDD关联schema
    val bdf = sqlContext.createDataFrame(rowRDD, structType)

    // 不使用SQL的方式，就不需要注册临时表
    val frame = bdf.select("id", "name", "score")


    import  sqlContext.implicits._
    val frame1 = bdf.orderBy($"score" desc, $"age" asc)

    frame1.show()

    sc.stop()
  }

}

在Spark2.x版本中，引入了SparkSession的概念，它为用户提供了一个统一的切入点来使用Spark的各项功能，用户不但可以使用DataFrame的各种API和可以使用SQL，学习Spark2的难度也会大大降低。SparkConf、SparkContext和SQLContext都已经被封装在SparkSession当中。

在该版本下，我选择使用的scala2.11.12版本

1、创建SparkSession,由于SparkContext被封装在SparkSession中，直接调用
SparkSession.SparkContext.textFile读取文件,在RDD中关联Row，将非结构化的数据转化为结构化的数据，然后定义模式，通过调用SqlContext的createDataFrame方法将RDD转化为DataFrame.

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * Spark2.x 编程模式
 * 1、创建SparkSession,SparkContext被封装在SparkSession中，直接调用读取文件
 * 2、先创建RDD，对数据进行整理，然后关联ROW，将非结构化数据转化为结构化数据
 * 3、定义schema
 * 4、调用SparkSession的createDataFrame方法
 * 5、注册临时表
 * 6、执行DataFrame语句（Transformation方式）
 * 7、执行action
 */
object SparkTest1 {
  def main(args: Array[String]): Unit = {

    //spark2.x SQL的编程API(SparkSession)
    //是spark2.x SQL执行的入口
    val session = SparkSession.builder()
      .appName("SQLTest1")
      .master("local[*]")
      .getOrCreate()

    //创建RDD
    val lines: RDD[String] = session.sparkContext.textFile("hdfs://master:9000/test/sparkSQL/person.txt")

    //将数据进行整理
    val rowRDD: RDD[Row] = lines.map(line => {
      val fields = line.split(",")
      val id = fields(0).toInt
      val name = fields(1)
      val age = fields(2).toInt
      val score = fields(3).toDouble
      Row(id, name, age, score)
    })

    //结果类型，其实就是表头，用于描述DataFrame
    val schema: StructType = StructType(List(
      StructField("id", IntegerType, true),
      StructField("name", StringType, true),
      StructField("age", IntegerType, true),
      StructField("score", DoubleType, true)
    ))

    //创建DataFrame
    val df: DataFrame = session.createDataFrame(rowRDD, schema)

    // 导入隐式转化
    import session.implicits._
    val df2: Dataset[Row] = df.where($"score" > 98).orderBy($"score" desc, $"age" asc)

    df2.show()

    session.stop()
  }

}

2、创建DataFrame后，同样也可以调用createTempView将DataFrame注册成视图，在视图里使用SQL进行查询。如上面的一样，省略。。。。

相关推荐

redis的八种使用场景

前言:redis是我们工作开发中，经常要打交道的，下面对redis的使用场景做总结介绍也是对redis举报的功能做梳理。缓存Redis最常见的用途是作为缓存，用于加速应用程序的响应速度。...

基于Redis的3种分布式ID生成策略

在分布式系统设计中，全局唯一ID是一个基础而关键的组件。随着业务规模扩大和系统架构向微服务演进，传统的单机自增ID已无法满足需求。高并发、高可用的分布式ID生成方案成为构建可靠分布式系统的必要条件。R...

基于OpenWrt系统路由器的模式切换与网页设计

摘要：目前商用WiFi路由器已应用到多个领域，商家通过给用户提供一个稳定免费WiFi热点达到吸引客户、提升服务的目标。传统路由器自带的Luci界面提供了工厂模式的Web界面，用户可通过该界面配置路...

这篇文章教你看明白 nginx-ingress 控制器

主机nginx一般nginx做主机反向代理（网关）有以下配置...

如何用redis实现注册中心

一句话总结使用Redis实现注册中心：服务注册...

爱可可老师24小时热门分享(2020.5.10)

No1.看自己以前写的代码是种什么体验？No2.DooM-chip!国外网友SylvainLefebvre自制的无CPU、无操作码、无指令计数器...No3.我认为CS学位可以更好，如...

Apportable:拯救程序员，IOS一秒变安卓

摘要：还在为了跨平台使用cocos2d-x吗,拯救objc程序员的奇葩来了,ApportableSDK:FreeAndroidsupportforcocos2d-iPhone。App...

JAVA实现超买超卖方案汇总，那个最适合你，一篇文章彻底讲透

以下是几种Java实现超买超卖问题的核心解决方案及代码示例，针对高并发场景下的库存扣减问题：方案一：Redis原子操作+Lua脚本（推荐）//使用Redis+Lua保证原子性publicbo...

3月26日更新 快速施法自动施法可独立设置

2016年3月26日DOTA2有一个79.6MB的更新主要是针对自动施法和快速施法的调整本来内容不多不少朋友都有自动施法和快速施法的困扰英文更新日志一些视觉BUG修复就不翻译了主要翻译自动施...

Redis 是如何提供服务的

在刚刚接触Redis的时候，最想要知道的是一个’setnameJhon’命令到达Redis服务器的时候，它是如何返回’OK’的？里面命令处理的流程如何，具体细节怎么样？你一定有问过自己...

lua _G、_VERSION使用

到这里我们已经把lua基础库中的函数介绍完了，除了函数外基础库中还有两个常量，一个是_G，另一个是_VERSION。_G是基础库本身，指向自己，这个变量很有意思，可以无限引用自己，最后得到的还是自己，...

China's top diplomat to chair third China-Pacific Island countries foreign ministers' meeting

BEIJING,May21(Xinhua)--ChineseForeignMinisterWangYi,alsoamemberofthePoliticalBureau...

移动工作交流工具Lua推出Insights数据分析产品

Lua是一个适用于各种职业人士的移动交流平台，它在今天推出了一项叫做Insights的全新功能。Insights是一个数据平台，客户可以在上面实时看到员工之间的交流情况，并分析这些情况对公司发展的影响...

Redis 7新武器:用Redis Stack实现向量搜索的极限压测

当传统关系型数据库还在为向量相似度搜索的性能挣扎时，Redis7的RedisStack...

Nginx/OpenResty详解，Nginx Lua编程，重定向与内部子请求

重定向与内部子请求Nginx的rewrite指令不仅可以在Nginx内部的server、location之间进行跳转，还可以进行外部链接的重定向。通过ngx_lua模块的Lua函数除了能实现Nginx...

一周热门
最近发表
标签列表
赣ICP备2024040476号-12