[TOC]
Spark SQL是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用。类似于hive的作用。
1、容易集成:安装Spark的时候,已经集成好了。不需要单独安装。
2、统一的数据访问方式:JDBC、JSON、Hive、parquet文件(一种列式存储文件,是SparkSQL默认的数据源,hive中也支持)
3、完全兼容Hive。可以将Hive中的数据,直接读取到Spark SQL中处理。
一般在生产中,基本都是使用hive做数据仓库存储数据,然后用spark从hive读取数据进行处理。
4、支持标准的数据连接:JDBC、ODBC
5、计算效率比基于mr的hive高,而且hive2.x版本中,hive建议使用spark作为执行引擎
DataFrame是组织成命名列的数据集。它在概念上等同于关系数据库中的表,里面有表的结构以及数据,但在底层具有更丰富的优化。DataFrames可以从各种来源构建,
例如:
结构化数据文件
hive中的表
外部数据库或现有RDDs
DataFrame API支持的语言有Scala,Java,Python和R。
比起RDD,DataFrame多了数据的结构信息,即schema。RDD是分布式的 Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外,更重要的特点是提升执行效率、减少数据读取以及执行计划的优化。
Dataset是一个分布式的数据收集器。这是在Spark1.6之后新加的一个接口,兼顾了RDD的优点(强类型,可以使用功能强大的lambda)以及Spark SQL的执行器高效性的优点。所以可以把DataFrames看成是一种特殊的Datasets,即:Dataset(Row)
Apache Spark 2.0引入了SparkSession,其为用户提供了一个统一的切入点来使用Spark的各项功能,并且允许用户通过它调用DataFrame和Dataset相关API来编写Spark程序。最重要的是,它减少了用户需要了解的一些概念,使得我们可以很容易地与Spark交互。
在2.0版本之前,与Spark交互之前必须先创建SparkConf和SparkContext。然而在Spark 2.0中,我们可以通过SparkSession来实现同样的功能,而不需要显式地创建SparkConf, SparkContext 以及 SQLContext,因为这些对象已经封装在SparkSession中。
要注意一点,在我用的这个spark版本中,直接使用new SQLContext() 来创建SQLContext对象,会显示该方式已经被弃用了(IDEA中会显示已弃用),建议使用SparkSession来获取SQLContext对象。
这种方式在scala中比较常用,因为case class是scala的特色
/**
表 t_stu 的结构为:
id name age
*/
object CreateDF {
def main(args: Array[String]): Unit = {
//这是最新的获取SQLContext对象的方式
//2、创建SparkSession对象,设置master,appname
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
//3、通过spark获取sparkContext对象,读取数据
val lines = spark.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))
//4、将数据映射到case class中,也就是数据映射到表的对应字段中
val tb = lines.map(t=>emp(t(0).toInt,t(1),t(2).toInt))
//这里必须要加上隐式转换,否则无法调用 toDF 函数
import spark.sqlContext.implicits._
//5、生成df
val df2 = tb.toDF()
//相当于select name from t_stu
df1.select($"name").show()
//关闭spark对象
spark.stop()
}
}
/*1、定义case class,每个属性对应表中的字段名以及类型
一般生产中为了方便,会全部定义为string类型,然后有需要的时候
才根据实际情况将string转为需要的类型
这一步相当于定义表的结构
*/
case class emp(id:Int,name:String,age:Int)
总结步骤为:
1、定义case class,用来表结构
2、创建sparkSession对象,用来读取数据
3、将rdd中的数据和case class映射
4、调用 toDF 函数将rdd转为 DataFrame
这种方式java比较常用
package SparkSQLExer
import org.apache
import org.apache.spark
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
/**
* 创建dataschema方式2:
* 通过spark session对象创建,表结构通过StructType创建
*/
object CreateDF02 {
def main(args: Array[String]): Unit = {
val sparkS = SparkSession.builder().master("local").appName("create schema").getOrCreate()
//1、通过StructType创建表结构schema,里面表的每个字段使用 StructField定义
val tbSchema = StructType(List(
StructField("id",DataTypes.IntegerType),
StructField("name",DataTypes.StringType),
StructField("age",DataTypes.IntegerType)
))
//2、读取数据
var lines = sparkS.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))
//3、将数据映射为ROW对象
val rdd1 = lines.map(t=>Row(t(0).toInt,t(1),t(2).toInt))
//4、创建表结构和表数据映射,返回的就是df
val df2 = sparkS.createDataFrame(rdd1, tbSchema)
//打印表结构
df2.printSchema()
sparkS.stop()
}
}
总结步骤为:
1、通过StructType创建表结构schema,里面表的每个字段使用 StructField定义
2、通过sparkSession.sparkContext读取数据
3、将数据映射格式为Row对象
4、将StructType和数据Row对象映射,返回df
package SparkSQLExer
import org.apache.spark.sql.SparkSession
/**
* 创建df方式3:通过有格式的文件直接导入数据以及表结构,比如json格式的文件
* 返回的直接就是一个DF
*/
object CreateDF03 {
def main(args: Array[String]): Unit = {
val sparkS = SparkSession.builder().master("local").appName("create df through json").getOrCreate()
//读取json方式1:
val jsonrdd1= sparkS.read.json("path")
//读取json方式2:
val jsonrdd1= sparkS.read.format("json").load("path")
sparkS.stop()
}
}
这种方式比较简单,就是直接读取json文件而已
sparkS.read.xxxx读取任意文件时,返回的都是DF对象
DSL语句其实就是将sql语句的一些操作转为类似函数的方式去调用,比如:
df1.select("name").show
例子:
为了方便,直接在spark-shell里操作了,
spark-shell --master spark://bigdata121:7077
1、打印表结构
scala> df1.printSchema
root
|-- empno: integer (nullable = true)
|-- ename: string (nullable = true)
|-- job: string (nullable = true)
|-- mgr: integer (nullable = true)
|-- hiredate: string (nullable = true)
|-- sal: integer (nullable = true)
|-- comm: integer (nullable = true)
|-- deptno: integer (nullable = true)
2、显示当前df的表数据或者查询结果的数据
scala> df1.show
+-----+------+---------+----+----------+----+----+------+
|empno| ename| job| mgr| hiredate| sal|comm|deptno|
+-----+------+---------+----+----------+----+----+------+
| 7369| SMITH| CLERK|7902|1980/12/17| 800| 0| 20|
| 7499| ALLEN| SALESMAN|7698| 1981/2/20|1600| 300| 30|
| 7521| WARD| SALESMAN|7698| 1981/2/22|1250| 500| 30|
| 7566| JONES| MANAGER|7839| 1981/4/2|2975| 0| 20|
| 7654|MARTIN| SALESMAN|7698| 1981/9/28|1250|1400| 30|
| 7698| BLAKE| MANAGER|7839| 1981/5/1|2850| 0| 30|
| 7782| CLARK| MANAGER|7839| 1981/6/9|2450| 0| 10|
| 7788| SCOTT| ANALYST|7566| 1987/4/19|3000| 0| 20|
| 7839| KING|PRESIDENT|7839|1981/11/17|5000| 0| 10|
| 7844|TURNER| SALESMAN|7698| 1981/9/8|1500| 0| 30|
| 7876| ADAMS| CLERK|7788| 1987/5/23|1100| 0| 20|
| 7900| JAMES| CLERK|7698| 1981/12/3| 950| 0| 30|
| 7902| FORD| ANALYST|7566| 1981/12/3|3000| 0| 20|
| 7934|MILLER| CLERK|7782| 1982/1/23|1300| 0| 10|
+-----+------+---------+----+----------+----+----+------+
3、执行select, 相当于select xxx form xxx where xxx
scala> df1.select("ename","sal").where("sal>2000").show
+------+----+
| ename| sal|
+------+----+
| SMITH| 800|
| ALLEN|1600|
| WARD|1250|
| JONES|2975|
|MARTIN|1250|
| BLAKE|2850|
| CLARK|2450|
| SCOTT|3000|
| KING|5000|
|TURNER|1500|
| ADAMS|1100|
| JAMES| 950|
| FORD|3000|
|MILLER|1300|
+------+----+
4、对某些列进行操作
对某个指定进行操作时,需要加上$符号,然后后面才能操作
$代表 取出来以后,再做一些操作。
注意:这个 $ 的用法在ideal中无法正常使用,解决方法下面说
scala> df1.select($"ename",$"sal",$"sal"+100).show
+------+----+-----------+
| ename| sal|(sal + 100)|
+------+----+-----------+
| SMITH| 800| 900|
| ALLEN|1600| 1700|
| WARD|1250| 1350|
| JONES|2975| 3075|
|MARTIN|1250| 1350|
| BLAKE|2850| 2950|
| CLARK|2450| 2550|
| SCOTT|3000| 3100|
| KING|5000| 5100|
|TURNER|1500| 1600|
| ADAMS|1100| 1200|
| JAMES| 950| 1050|
| FORD|3000| 3100|
|MILLER|1300| 1400|
+------+----+-----------+
5、过滤行
scala> df1.filter($"sal">2000).show
+-----+-----+---------+----+----------+----+----+------+
|empno|ename| job| mgr| hiredate| sal|comm|deptno|
+-----+-----+---------+----+----------+----+----+------+
| 7566|JONES| MANAGER|7839| 1981/4/2|2975| 0| 20|
| 7698|BLAKE| MANAGER|7839| 1981/5/1|2850| 0| 30|
| 7782|CLARK| MANAGER|7839| 1981/6/9|2450| 0| 10|
| 7788|SCOTT| ANALYST|7566| 1987/4/19|3000| 0| 20|
| 7839| KING|PRESIDENT|7839|1981/11/17|5000| 0| 10|
| 7902| FORD| ANALYST|7566| 1981/12/3|3000| 0| 20|
+-----+-----+---------+----+----------+----+----+------+
6、分组以及计数
scala> df1.groupBy($"deptno").count.show
+------+-----+
|deptno|count|
+------+-----+
| 20| 5|
| 10| 3|
| 30| 6|
+------+-----+
上面说到在ide中 select($"name")中无法正常使用,解决方法为:
在该语句之前加上这么一句:
import spark.sqlContext.implicits._
主要还是因为类型的问题,加上隐式转换就好了
df对象不能直接执行sql。需要生成一个视图,再执行SQL。
需要指定创建的视图名称,后面视图名称就相当于表名。
视图后面还会细说,这里先有个概念
例子:
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
。。。。。。。。。。。。。。
//通过df对象创建临时视图。视图名就相当于表名
df1.createOrReplaceTempView("emp")
//通过sparksession对象执行执行
spark.sql("select * from emp").show
spark.sql("select * from emp where sal > 2000").show
spark.sql("select deptno,count(1) from emp group by deptno").show
//可以创建多个视图,不冲突
df1.createOrReplaceTempView("emp12345")
spark.sql("select e.deptno from emp12345 e").show
scala> case class Dept(deptno:Int,dname:String,loc:String)
defined class Dept
scala> val lines = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
lines: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[68] at map at <console>:24
scala> val allDept = lines.map(x=>Dept(x(0).toInt,x(1),x(2)))
allDept: org.apache.spark.rdd.RDD[Dept] = MapPartitionsRDD[69] at map at <console>:28
scala> val df2 = allDept.toDF
df2: org.apache.spark.sql.DataFrame = [deptno: int, dname: string ... 1 more field]
scala> df2.create
createGlobalTempView createOrReplaceTempView createTempView
scala> df2.createOrReplaceTempView("dept")
scala> spark.sql("select dname,ename from emp12345,dept where emp12345.deptno=dept.deptno").show
+----------+------+
| dname| ename|
+----------+------+
| RESEARCH| SMITH|
| RESEARCH| JONES|
| RESEARCH| SCOTT|
| RESEARCH| ADAMS|
| RESEARCH| FORD|
|ACCOUNTING| CLARK|
|ACCOUNTING| KING|
|ACCOUNTING|MILLER|
| SALES| ALLEN|
| SALES| WARD|
| SALES|MARTIN|
| SALES| BLAKE|
| SALES|TURNER|
| SALES| JAMES|
+----------+------+
和DataFrame类似,只是把 toDF改为调用 toDS 方法
package SparkSQLExer
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
object CreateDS {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
val lines = spark.sparkContext.textFile("G:\\test\\t_stu.txt").map(_.split(","))
val tb = lines.map(t=>emp1(t(0).toInt,t(1),t(2).toInt))
import spark.sqlContext.implicits._
val df1 = tb.toDS()
df1.select($"name")
}
}
case class emp1(id:Int,name:String,age:Int)
package SparkSQLExer
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
object CreateDS {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
//创建一个序列对象,里面都是emp1对象,映射的数据,然后直接toDS转为DataSet
val ds1 = Seq(emp1(1,"king",20)).toDS()
ds1.printSchema()
}
}
case class emp1(id:Int,name:String,age:Int)
定义case class
case class Person(name:String,age:BigInt)
使用JSON数据生成DataFrame
val df = spark.read.format("json").load("/usr/local/tmp_files/people.json")
将DataFrame转换成DataSet
df.as[Person].show
df.as[Person] 是一个 DataSet
as[T]中的泛型需要是一个case class类,用于映射表头
DataSet支持的算子其实就是rdd和DataFrame算子的结合。
使用emp.json 生成DataFrame
val empDF = spark.read.json("/usr/local/tmp_files/emp.json")
scala> empDF.show
+----+------+-----+------+----------+---------+----+----+
|comm|deptno|empno| ename| hiredate| job| mgr| sal|
+----+------+-----+------+----------+---------+----+----+
| | 20| 7369| SMITH|1980/12/17| CLERK|7902| 800|
| 300| 30| 7499| ALLEN| 1981/2/20| SALESMAN|7698|1600|
| 500| 30| 7521| WARD| 1981/2/22| SALESMAN|7698|1250|
| | 20| 7566| JONES| 1981/4/2| MANAGER|7839|2975|
|1400| 30| 7654|MARTIN| 1981/9/28| SALESMAN|7698|1250|
| | 30| 7698| BLAKE| 1981/5/1| MANAGER|7839|2850|
| | 10| 7782| CLARK| 1981/6/9| MANAGER|7839|2450|
| | 20| 7788| SCOTT| 1987/4/19| ANALYST|7566|3000|
| | 10| 7839| KING|1981/11/17|PRESIDENT| |5000|
| 0| 30| 7844|TURNER| 1981/9/8| SALESMAN|7698|1500|
| | 20| 7876| ADAMS| 1987/5/23| CLERK|7788|1100|
| | 30| 7900| JAMES| 1981/12/3| CLERK|7698| 950|
| | 20| 7902| FORD| 1981/12/3| ANALYST|7566|3000|
| | 10| 7934|MILLER| 1982/1/23| CLERK|7782|1300|
+----+------+-----+------+----------+---------+----+----+
scala> empDF.where($"sal" >= 3000).show
+----+------+-----+-----+----------+---------+----+----+
|comm|deptno|empno|ename| hiredate| job| mgr| sal|
+----+------+-----+-----+----------+---------+----+----+
| | 20| 7788|SCOTT| 1987/4/19| ANALYST|7566|3000|
| | 10| 7839| KING|1981/11/17|PRESIDENT| |5000|
| | 20| 7902| FORD| 1981/12/3| ANALYST|7566|3000|
+----+------+-----+-----+----------+---------+----+----+
#### empDF 转换成 DataSet 需要 case class
scala> case class Emp(empno:BigInt,ename:String,job:String,mgr:String,hiredate:String,sal:BigInt,comm:String,deptno:BigInt)
defined class Emp
scala> val empDS = empDF.as[Emp]
empDS: org.apache.spark.sql.Dataset[Emp] = [comm: string, deptno: bigint ... 6 more fields]
scala> empDS.filter(_.sal > 3000).show
+----+------+-----+-----+----------+---------+---+----+
|comm|deptno|empno|ename| hiredate| job|mgr| sal|
+----+------+-----+-----+----------+---------+---+----+
| | 10| 7839| KING|1981/11/17|PRESIDENT| |5000|
+----+------+-----+-----+----------+---------+---+----+
scala> empDS.filter(_.deptno == 10).show
+----+------+-----+------+----------+---------+----+----+
|comm|deptno|empno| ename| hiredate| job| mgr| sal|
+----+------+-----+------+----------+---------+----+----+
| | 10| 7782| CLARK| 1981/6/9| MANAGER|7839|2450|
| | 10| 7839| KING|1981/11/17|PRESIDENT| |5000|
| | 10| 7934|MILLER| 1982/1/23| CLERK|7782|1300|
+----+------+-----+------+----------+---------+----+----+
多表查询:
1、创建部门表
scala> val deptRDD = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
deptRDD: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[154] at map at <console>:24
scala> case class Dept(deptno:Int,dname:String,loc:String)
defined class Dept
scala> val deptDS = deptRDD.map(x=>Dept(x(0).toInt,x(1),x(2))).toDS
deptDS: org.apache.spark.sql.Dataset[Dept] = [deptno: int, dname: string ... 1 more field]
scala> deptDS.show
+------+----------+--------+
|deptno| dname| loc|
+------+----------+--------+
| 10|ACCOUNTING|NEW YORK|
| 20| RESEARCH| DALLAS|
| 30| SALES| CHICAGO|
| 40|OPERATIONS| BOSTON|
+------+----------+--------+
2、员工表
同上 empDS
empDS.join(deptDS,"deptno").where(xxxx) 连接两个表,通过deptno字段
empDS.joinWith(deptDS,deptDS("deptno")===empDS("deptno")) 这个用于连接的字段名称不一样的情况
如果想使用标准的sql语句来操作df或者ds对象时,必须先给df或者ds对象创建视图,然后通过SparkSession对象的sql函数来对相应的视图进行操作才可以。那么视图是什么?
视图是一个虚表,不存储数据,可以当做是表的一个访问链接。视图有两种类型:
普通视图:也叫本地视图,只在当前session会话中有效
全局视图:在全部session中都有效,全局视图创建在指定命名空间中:global_temp 类似于一个库
操作说明:
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
val empDF = spark.read.json("/usr/local/tmp_files/emp.json")
创建本地视图:
empDF.createOrReplaceTempView(视图名),视图存在就会重新创建
empDF.createTempView(视图名),如果视图存在就不会创建
创建全局视图:
empDF.createGlobalTempView(视图名)
对视图执行sql操作,这里视图名就类似于表名
spark.sql("xxxxx")
例子:
empDF.createOrReplaceTempView("emp")
spark.sql("select * from emp").show
注意,只要创建了视图,那么就可以通过sparksession对象在任意一个类中操作视图,也就是表。这个特性很好用,当我们要操作一些表时,可以一开始就读取成df,然后创建成视图,那么就可以在任意一个地方查询表了。
通过SparkSession对象可以读取不同格式的数据源:
val spark = SparkSession.builder().master("local").appName("createDF case class").getOrCreate()
下面都用上面的spark代称SparkSession。
1、load
spark.read.load(path):读取指定路径的文件,要求文件存储格式为Parquet文件
2、format
spark.read.format("格式").load(path) :指定读取其他格式的文件,如json
例子:
spark.read.format("json").load(path)
3、直接读取其他格式文件
spark.read.格式名(路径),这是上面2中的一个简写方式,例子:
spark.read.json(路径) json格式文件
spark.read.text(路径) 读取文本文件
注意:这些方式返回的都是 DataFrame 对象
可以将DataFrame 对象写入到指定格式的文件中,假设有个DataFrame 对象为df1.
1、save
df1.write.save(路径)
他会将文件保存到这个目录下,文件名spark随机生成的,所以使用上面的读取方式的时候,直接指定读取目录即可,不用指定文件名。输出的文件格式为 Parquet。可以直接指定hdfs的路径,否则就存储到本地
如:
df1.write.save("/test")
spark.read.load("/test")
2、直接指定格式存储
df1.write.json(路径) 这样就会以json格式保存文件,生成的文件名的情况和上面类似
3、指定保存模式
如果没有指定保存模式,输出路径存在的情况下,就会报错
df1.write.mode("append").json(路径)
mode("append") 就表示文件存在时就追加
mode("overwrite") 表示覆盖旧数据
4、保存为表
df1.write.saveAsTable(表名) 会保存在当前目录的spark-warehouse 目录下
5、format
df1.write.format(格式).save()
使用指定特定格式的方式来输出保存数据,比如保存到MongoDB数据库中
这种一种列式存储格式,具体原理可以看看之前hive的文章。这种格式是默认的存储格式,使用load和save时默认的格式,操作方式很像前面说的,这里不重复。这里要讲的是Parquet的一个特殊的功能,支持schema(表结构)的合并。例子:
scala> val df1 = sc.makeRDD(1 to 5).map(i=>(i,i*2)).toDF("single","double")
df1: org.apache.spark.sql.DataFrame = [single: int, double: int]
scala> df1.show
+------+------+
|single|double|
+------+------+
| 1| 2|
| 2| 4|
| 3| 6|
| 4| 8|
| 5| 10|
+------+------+
scala> sc.makeRDD(1 to 5)
res8: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[26] at makeRDD at <console>:25
scala> sc.makeRDD(1 to 5).collect
res9: Array[Int] = Array(1, 2, 3, 4, 5)
//导出表1
scala> df1.write.parquet("/usr/local/tmp_files/test_table/key=1")
scala> val df2 = sc.makeRDD(6 to 10).map(i=>(i,i*3)).toDF("single","triple")
df2: org.apache.spark.sql.DataFrame = [single: int, triple: int]
scala> df2.show
+------+------+
|single|triple|
+------+------+
| 6| 18|
| 7| 21|
| 8| 24|
| 9| 27|
| 10| 30|
+------+------+
//导出表2
scala> df2.write.parquet("/usr/local/tmp_files/test_table/key=2")
scala> val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")
df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 1 more field]
//直接读取会丢失字段
scala> df3.show
+------+------+---+
|single|double|key|
+------+------+---+
| 8| null| 2|
| 9| null| 2|
| 10| null| 2|
| 3| 6| 1|
| 4| 8| 1|
| 5| 10| 1|
| 6| null| 2|
| 7| null| 2|
| 1| 2| 1|
| 2| 4| 1|
+------+------+---+
//加上option,指定"mergeSchema"为true,就可以合并
scala> val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")
df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 2 more fields]
scala> df3.show
+------+------+------+---+
|single|double|triple|key|
+------+------+------+---+
| 8| null| 24| 2|
| 9| null| 27| 2|
| 10| null| 30| 2|
| 3| 6| null| 1|
| 4| 8| null| 1|
| 5| 10| null| 1|
| 6| null| 18| 2|
| 7| null| 21| 2|
| 1| 2| null| 1|
| 2| 4| null| 1|
+------+------+------+---+
补充问题:key 是什么?必须用key嘛?
key是不同表的一个区分字段,在合并的时候,会作为合并之后的表的一个字段,并且值等于key=xx 中设置的值
如果目录下,两个表的目录名不一样,是无法合并的,合并字段名可以任意,
如:一个是key ,一个是 test 这两个无法合并,必须统一key或者test
这种一种带表格式字段的文件,例子:
scala> val peopleDF = spark.read.json("/usr/local/tmp_files/people.json")
peopleDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
scala> peopleDF.printSchema()
root
|-- age: long (nullable = true)
|-- name: string (nullable = true)
scala> peopleDF.createOrReplaceTempView("people")
scala> spark.sql("select * from people where age=19")
res25: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
scala> spark.sql("select * from people where age=19").show
+---+------+
|age| name|
+---+------+
| 19|Justin|
+---+------+
scala> spark.sql("select age,count(1) from people group by age").show
+----+--------+
| age|count(1)|
+----+--------+
| 19| 1|
|null| 1|
| 30| 1|
+----+--------+
df对象支持通过jdbc连接数据库,写入数据到数据库,或者从数据库读取数据。
例子:
1、通过jdbc 从mysql读取数据:
使用 format(xx).option()的方式指定连接数据库的一些参数,比如用户名密码,使用的连接驱动等
import java.util.Properties
import org.apache.spark.sql.SparkSession
object ConnMysql {
def main(args: Array[String]): Unit = {
val sparkS = SparkSession.builder().appName("spark sql conn mysql").master("local").getOrCreate()
//连接mysql方式1:
//创建properties配置对象,用于存放连接mysql的参数
val mysqlConn = new Properties()
mysqlConn.setProperty("user","root")
mysqlConn.setProperty("password","wjt86912572")
//使用jdbc连接,指定连接字符串,表名,以及其他连接参数,并返回对应的dataframe
val mysqlDF1 = sparkS.read.jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8","customer",mysqlConn)
mysqlDF1.printSchema()
mysqlDF1.show()
mysqlDF1.createTempView("customer")
sparkS.sql("select * from customer limit 2").show()
//连接mysql方式2,这种方式比较常用:
val mysqlConn2 = sparkS.read.format("jdbc")
.option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8")
.option("user","root")
.option("password","wjt86912572")
.option("driver","com.mysql.jdbc.Driver")
.option("dbtable","customer").load()
mysqlConn2.printSchema()
}
}
这是两种连接读取数据的方式。
2、jdbc写入数据到mysql
和读取类似,只不过换成了write操作
import java.util.Properties
import org.apache.spark.sql.SparkSession
object WriteToMysql {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("write to mysql").master("local").getOrCreate()
val df1 = spark.read.text("G:\\test\\t_stu.json")
//方式1:
df1.write.format("jdbc")
.option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8")
.option("user","root")
.option("password","wjt86912572")
.option("driver","com.mysql.jdbc.Driver")
.option("dbtable","customer").save()
//方式2:
val mysqlConn = new Properties()
mysqlConn.setProperty("user","root")
mysqlConn.setProperty("password","wjt86912572")
df1.write.jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8","customer",mysqlConn)
}
}
必须要保证df的表格式和写入的mysql的表格式一样,字段名也要一样
1、通过jdbc连接hive
方式和普通jdbc类似,例如:
import java.util.Properties
import org.apache.spark.sql.SparkSession
/**
* 连接hive的情况有两种:
* 1、如果是直接在ideal中运行spark程序的话,则必须在程序中指定jdbc连接的hiveserver的地址
* 且hiveserver必须以后台服务的形式暴露10000端口出来.这种方式是直接通过jdbc连接hive
*
* 2、如果程序是打包到spark集群中运行的话,一般spark集群的conf目录下,已经有hive client
* 的配置文件,就会直接启动hive client来连接hive。这时不需要启动hiveserver服务。
* 这种方式是通过hive client连接hive
*/
object ConnHive {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("spark sql conn mysql").master("local").getOrCreate()
val properties = new Properties()
properties.setProperty("user","")
properties.setProperty("password","")
val hiveDF = spark.read.jdbc("jdbc:hive2://bigdata121:10000/default","customer",properties)
hiveDF.printSchema()
spark.stop()
}
}
这种方式要注意一点:
hiveserver必须以后台服务的形式暴露10000端口出来.这种方式是直接通过jdbc连接hive。
2、通过hive client连接hive
这种方式一般用在生产中,因为任务一般都是通过spark-submit提交到集群中运行,这时候就会直接通过hive client来连接hive,不会通过jdbc来连接了。
要注意:需要在spark的节点上都配置上hive client,然后将hive-site.xml配置文件拷贝到 spark的conf目录下。同时需要将hadoop的core-site.xml hdfs-site.xml也拷贝过去。另外一方面,因为要使用hive client,所以hive server那边,一般都要配置metastore server,具体配置看hive的文章。
这样在spark集群中的程序就可以直接使用
spark.sql("xxxx").show
这样的操作,默认就会从hive中读取对应的表进行操作。不用另外做任何连接hive 的操作
或者直接到 spark-shell中,也是可以直接使用 上面的方式操作hive的表
例如:
import org.apache.spark.sql.SparkSession
object ConnHive02 {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("spark sql conn hive").getOrCreate()
spark.sql("select * from customer").show()
}
}
这样直接操作的就是 hive 的表了
import java.util.Properties
import org.apache.spark.sql.SparkSession
object HiveToMysql {
def main(args: Array[String]): Unit = {
//直接通过spark集群中的hive client连接hive,不需要jdbc以及hive server
val spark = SparkSession.builder().appName("hive to mysql").enableHiveSupport().getOrCreate()
val resultDF = spark.sql("select * from default.customer")
//一般中间写的处理逻辑都是处理从hive读取的数据,处理完成后写入到mysql
val mysqlConn = new Properties()
mysqlConn.setProperty("user","root")
mysqlConn.setProperty("password","wjt86912572")
//通过jdbc写入mysql
resultDF.write.mode("append").jdbc("jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8", "customer", mysqlConn)
spark.stop()
}
}
先启动个spark-shell
spark-shell --master spark://bigdata121:7077
要在spark-shell中操作mysql,所以记得自己找个 mysql-connector的jar放到spark的jars目录下
例子:
创建df,从mysql读取表
scala> val mysqDF = spark.read.format("jdbc").option("url","jdbc:mysql://bigdata121:3306/test?serverTimezone=UTC&characterEncoding=utf-8").option("user","root").option("password","wjt86912572").option("driver","com.mysql.jdbc.Driver").option("dbtable","customer").load()
mysqDF: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
scala> mysqDF.show
+---+------+--------------------+
| id| name| last_mod|
+---+------+--------------------+
| 1| neil|2019-07-20 17:09:...|
| 2| jack|2019-07-20 17:09:...|
| 3|martin|2019-07-20 17:09:...|
| 4| tony|2019-07-20 17:09:...|
| 5| eric|2019-07-20 17:09:...|
| 6| king|2019-07-20 17:42:...|
| 7| tao|2019-07-20 17:45:...|
+---+------+--------------------+
必须注册成一张表,才可以缓存。
scala> mysqDF.registerTempTable("customer")
warning: there was one deprecation warning; re-run with -deprecation for details
标识这张表可以被缓存,但是现在数据并没有直接缓存
scala> spark.sqlContext.cacheTable("customer")
第一次查询表,从mysql读取数据,并缓存到内存中
scala> spark.sql("select * from customer").show
+---+------+--------------------+
| id| name| last_mod|
+---+------+--------------------+
| 1| neil|2019-07-20 17:09:...|
| 2| jack|2019-07-20 17:09:...|
| 3|martin|2019-07-20 17:09:...|
| 4| tony|2019-07-20 17:09:...|
| 5| eric|2019-07-20 17:09:...|
| 6| king|2019-07-20 17:42:...|
| 7| tao|2019-07-20 17:45:...|
+---+------+--------------------+
这一次查询从内存中返回
scala> spark.sql("select * from customer").show
+---+------+--------------------+
| id| name| last_mod|
+---+------+--------------------+
| 1| neil|2019-07-20 17:09:...|
| 2| jack|2019-07-20 17:09:...|
| 3|martin|2019-07-20 17:09:...|
| 4| tony|2019-07-20 17:09:...|
| 5| eric|2019-07-20 17:09:...|
| 6| king|2019-07-20 17:42:...|
| 7| tao|2019-07-20 17:45:...|
+---+------+--------------------+
清空缓存
scala> spark.sqlContext.clearCache
将数据缓存到内存中的相关优化参数
spark.sql.inMemoryColumnarStorage.compressed
默认为 true
Spark SQL 将会基于统计信息自动地为每一列选择一种压缩编码方式。
spark.sql.inMemoryColumnarStorage.batchSize
默认值:10000
缓存批处理大小。缓存数据时, 较大的批处理大小可以提高内存利用率和压缩率,但同时也会带来 OOM(Out Of Memory)的风险。
其他性能相关的配置选项(不过不推荐手动修改,可能在后续版本自动的自适应修改)
spark.sql.files.maxPartitionBytes
默认值:128 MB
读取文件时单个分区可容纳的最大字节数
spark.sql.files.openCostInBytes
默认值:4M
打开文件的估算成本, 按照同一时间能够扫描的字节数来测量。当往一个分区写入多个文件的时候会使用。高估更好, 这样的话小文件分区将比大文件分区更快 (先被调度)。
spark.sql.autoBroadcastJoinThreshold
默认值:10M
用于配置一个表在执行 join 操作时能够广播给所有 worker 节点的最大字节大小。通过将这个值设置为 -1 可以禁用广播。注意,当前数据统计仅支持已经运行了 ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan 命令的 Hive Metastore 表。
spark.sql.shuffle.partitions
默认值:200
用于配置 join 或聚合操作混洗(shuffle)数据时使用的分区数。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。