Spark SQL

2018-01-08

一丶目标

掌握sparksql原理
掌握DataFrame和DataSet的数据结构和使用方式
使用spark sql进行相应的编程

shark

基于spark的大数据规模底层数据仓库
依赖于hive底层代码,依赖于spark的版本
随着hive和spark版本的升级,对于shark的优化和维护带来了很大的影响

二丶 Spark SQL

1. spark SQL介绍

概述

Spark SQL 是spark用来处理结构化数据的一个模块,提供了一个编程抽象DataFrame,并且可以做为分布式SQL查询引擎的作用
使用: SQL丶 DataFrame API丶 Datasets API

特性

Intergrated (易整合)
- 可以使用sql,dataFrame,dataSet来操作
- 也可以使用4种语言来编程(java/scala/phthon/R)
Uniform Data Access (统一的数据访问方式)
Hive Intergration (兼容 Hive)
Standard Connectivity (标准的数据库链接)

2. DataFrame

概述

DataFrame是sparksql的底层抽象.可以理解为关系数据库中的二维表格,具有schema描述信息.spark1.3引入

在spark中,DataFrame是一种以RDD为基础的分布式数据集

DataFrame与RDD的优缺点比较

RDD

优点
1. 编译时类型安全
2. 面相对象的编程风格
缺点
1. 序列化和反序列化的性能开销大 — 无论是集群间的通信或者是IO操作,都会进行数据和结构的序列化和反序列化操作
2. GC的性能开销 — 频繁的创建和销毁对象

DataFrame

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

DataFrame创建方式

准备数据

person.txt
    1 zhangsan 20
    2 lisi 29
    3 wangwu 25
    4 zhaoliu 30
    5 tianqi 35
    5 kobe 40
上传到hdfs
	hdfs dfs -put person.txt /

读取文本文件创建DataFrame

val rdd1 = sc.textFile("person.txt").map(_.split(" "))

case class People(id:Int,name:String,age:Int)

//rdd1 与样例类关联
val rdd2 = rdd1.map(x=>People(x(0).toInt,x(1),x(2).toInt))

// 将RDD转换成DataFrame
val dataFrame = rdd2.toDF

//DataFrame处理
dataFrame.show
dataFrame.printSchema

//第二种方式,利用sparkSession 
val df = spark.read.text("xx.txt").show()

读取jaon文件创建DataFrame
1
spark.read.json("xx.json").show()

读取 parquet文件创建DataFrame

1	spark.read.parquet("xx.parquet").show()

Spark session创建
1
spark.createDataSpark()

3. DataFrame常用操作

DSL : 特定领域的语言
SQL :
- 需要将dataframe注册成一张表 registerTemple
- spark.sql(“sql语句”)

4. Dataset

4.1 概述

分布式数据集合,在spark1.6中添加的新的接口
集中了RDD的优点(强类型和可以用lambda函数)

4.2 DataFrame丶Dataset丶RDD的区别

Dataset包含了DataFrame的功能,在spark2.0 中两者统一,DataFrame表示为Dataset[Row],即DataSet的子集

DataSet可以在编译时检查
面相对象的编程接口

4.3 DataFrame和Dataset相互转换

DataFrame转为DataSet

df.as[ElementType]
Dataset转为DataFrame

ds.toDF()

4.4 创建Dataset

通过spark.createDataset创建
通toDS方法生成DataSet
通过DataFrame转换

三丶以编程的方式执行Spark SQL查询

1.

通过反射机制

//todo: 利用sparkSQL来创建DataFrame---反射机制(case class)
// 1丶创建sparkSession
val spark: SparkSession = SparkSession.builder().appName("xxx").master("local[2]").getOrCreate()

// 2丶获取sparkContext对象
val sc: SparkContext = spark.sparkContext

// 3丶读取数据文件
val dataRDD :RDD[Array[String]] = sc.textFile("xxx").map(_.split(" "))

// 4丶将RDD与样例类进行关联
val peopleRDD : RDD[People] = dataRDD.map(x=>People(x(0).toInt,x(1),x(2).toInt))

// 5丶将RDD转换为DataFrame
// 手动导入隐式转换
import spark.implicits._
val peopleDF : DataFrame = peopleRDD.toDF()

// 6丶打印schema
peopleDF.printSchema()
peopleDF.show() // 默认返回20条记录, boolean:表示当前超过20字符列名称的结果进行一个截取 默认返回true

// --------------------DSL语法 ---------------------
peopleDF.select("name").show() //查询一列
peopleDF.filter($"age" > 30)  //条件过滤
peopleDF.groupBy("age").count().show()  // 分组统计

// --------------------SQL 语法 --------------------
// 将DataFrame注册成一张临时表
peopleDF.createTempView("t_people")
// sparkSession.sql(sql)
spark.sql("select * from t_people").show()

// 7丶关闭
sc.stop()
spark.stop()


// 样例类
case class People(id:Int,name:String,age:Int)

通过StructType直接指定Schema

将RDD转换为包含ROW对象的RDD
基于structType类型创建schema,与第一步创建的RDD相匹配
通过sparkSession的createDataFrame

//todo: 利用sparkSql 将RDD转换为DataFrame --- 通过StructType直接指定Schema
// 1丶创建sparkSession
val spark: SparkSession = SparkSession.builder().appName("xxx").master("local[2]").getOrCreate()

// 2丶获取sparkContext对象
val sc: SparkContext = spark.sparkContext

// 3丶读取数据文件
val dataRDD :RDD[Array[String]] = sc.textFile("xxx").map(_.split(" "))

// 4丶 创建rowRDD
val rowRDD : RDD[Row] = dataRDD.map(x=>Row(x(0).toInt,x(1),x(2).toInt))

// 5丶指定DataFrame中的schema,用到StructType对象
val schema = StructType(StructField("id",IntegerType,false)::StructField("name",StringType,false)::StructField("age",IntegerType,false)::Nil)

// 6丶调用createDataFrame方法,生成DataFrame
val df : DataFrame = spark.createDataFrame(rowRDD,structType)

// 7丶打印schema
peopleDF.printSchema()
peopleDF.show() 

// 关闭
sc.stop()
spark.stop()

2. 利用sparkSQL操作HIVE SQL

val warehouseLocation = "D:\\workSpace_IDEA_NEW\\day2017-10-12\\spark-warehouse"
     //todo:1、创建sparkSession
    val spark: SparkSession = SparkSession.builder()
     									.appName("HiveSupport")
     			 						.master("local[2]")
     									.config("spark.sql.warehouse.dir", warehouseLocation)
     		 							.enableHiveSupport() //开启支持hive
     		 							.getOrCreate()
   spark.sparkContext.setLogLevel("WARN")  //设置日志输出级别
   import spark.implicits._
   import spark.sql

   //todo:2、操作sql语句
   sql("CREATE TABLE IF NOT EXISTS person (id int, name string, age int) row format delimited fields terminated by ' '")
   sql("LOAD DATA LOCAL INPATH '/person.txt' INTO TABLE person")
   sql("select * from person ").show()
   spark.stop()

四丶数据源

1. sparkSQL从mysql中加载数据

//todo:1、创建sparkSession对象
      val spark: SparkSession = SparkSession.builder()
        .appName("DataFromMysql")
        .master("local[2]")
        .getOrCreate()
//todo:2、创建Properties对象，设置连接mysql的用户名和密码
val properties: Properties =new Properties()
properties.setProperty("user","root")
properties.setProperty("password","123456")
//todo:3、读取mysql中的数据
val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://192.168.200.150:3306/spark","iplocaltion",properties)
//todo:4、显示mysql中表的数据
mysqlDF.show()
spark.stop()

2. sparkSQL将数据写入到mysql中

//todo:1、创建sparkSession对象
val spark: SparkSession = SparkSession.builder()
        .appName("SparkSqlToMysql")
        .getOrCreate()
//todo:2、读取数据
val data: RDD[String] = spark.sparkContext.textFile(args(0))
//todo:3、切分每一行,
val arrRDD: RDD[Array[String]] = data.map(_.split(" "))
//todo:4、RDD关联Student
val studentRDD: RDD[Student] = arrRDD.map(x=>Student(x(0).toInt,x(1),x(2).toInt))
//todo:导入隐式转换
import spark.implicits._
//todo:5、将RDD转换成DataFrame
val studentDF: DataFrame = studentRDD.toDF()
//todo:6、将DataFrame注册成表
studentDF.createOrReplaceTempView("student")
//todo:7、操作student表 ,按照年龄进行降序排列
val resultDF: DataFrame = spark.sql("select * from student order by age desc")

//todo:8、把结果保存在mysql表中
//todo:创建Properties对象，配置连接mysql的用户名和密码
val prop =new Properties()
prop.setProperty("user","root")
prop.setProperty("password","123456")

resultDF.write.jdbc("jdbc:mysql://192.168.200.150:3306/spark","student",prop)

//todo:写入mysql时，可以配置插入mode，overwrite覆盖，append追加，ignore忽略，error默认表存在报错
   //resultDF.write.mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.200.150:3306/spark","student",prop)
    spark.stop()
  }
}
//todo:创建样例类Student
case class Student(id:Int,name:String,age:Int)

一丶目标

二丶 Spark SQL

1. spark SQL介绍

2. DataFrame

DataFrame与RDD的优缺点比较

DataFrame创建方式

3. DataFrame常用操作

4. Dataset

4.1 概述

4.2 DataFrame丶Dataset丶RDD的区别

4.3 DataFrame和Dataset相互转换

4.4 创建Dataset

三丶 以编程的方式执行Spark SQL查询

1.

2. 利用sparkSQL操作HIVE SQL

四丶数据源

1. sparkSQL从mysql中加载数据

2. sparkSQL将数据写入到mysql中

三丶以编程的方式执行Spark SQL查询