SparkSQL |
您所在的位置：网站首页 › sparksql和rdd的区别 › SparkSQL |
SparkSQL
|
原文地址：https://www.toutiao.com/i6848607197815898628/ 本文主要从以下几个方面介绍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查询引擎。 第二，什么是DataFrameDataFrame的前身是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转化为DdtaFrame 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进行查询。如上面的一样，省略。。。。
|
【本文地址】
今日新闻 |
推荐新闻 |