码迷,mamicode.com
首页 > 数据库 > 详细

详细解读Spark的数据分析引擎:Spark SQL

时间:2018-03-04 19:01:51      阅读:231      评论:0      收藏:0      [点我收藏+]

标签:保存   overwrite   worker   ase   body   兼容   form   result   print   

欢迎关注大数据和人工智能技术文章发布的微信公众号:清研学堂,在这里你可以学到夜白(作者笔名)精心整理的笔记,让我们每天进步一点点,让优秀成为一种习惯!

 

技术分享图片

一、spark SQL:类似于Hive,是一种数据分析引擎

什么是spark SQL?

spark SQL只能处理结构化数据

底层依赖RDD,把sql语句转换成一个个RDD,运行在不同的worker上

特点:

1、容易集成:SQL语句

2、对不同的数据源提供统一的访问方式:DataFrame 用DataFrame屏蔽数据源的差别

3、兼容Hive

大纲:

技术分享图片

核心概念:DataFrame(看作表):就是表,是Spark SQL对结构化数据的抽象集合

表现形式:RDD

表=表结构+数据

DataFrame=schema+RDD

DataSet(新API接口 看作表)

如何创建DataFrame?

1、方式一:通过case class创建DataFrame

创建表结构

case class EMP(empno:Int,ename:String,job:String,mgr:String,hiredata:String,sal:Int,comm:String,deptno:Int)

导入emp.csv文件并指定分隔符

val lines = sc.textFile("/root/temp/emp.csv").map(_.split(","))

lines.collect

将表结构和数据关联起来

val allEmp = lines.map(x=>Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt))

创建DataFrame:

val empDF = allEmp.toDF

操作DataFrame:

empDF.show:展示DataFrame

empDF.printSchema:打印DataFrame的表结构

2、方式二:通过SparkSession.createDataFrame()创建DataFrame

什么是spark session?

从spark2.0以后提供了统一访问spark各个模块的对象:spark session

创建表结构:用StructType类

import org.apache.spark.sql

import org.apache.spark.sql.types._

val myschema = StructType(List(empno:Int,ename:String,job:String,mgr:String,hiredata:String,sal:Int,comm:String,deptno:Int))

导入emp.csv文件并指定分隔符

val lines = sc.textFile("/root/temp/emp.csv").map(_.split(","))

将表结构和数据关联起来,把读入的数据emp.csv映射成一行,这里没有带表结构

import.org.apache.spark.sql._

val rowRDD = lines.map(x=>Row(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt))

通过SparkSession.createDataFrame()创建表

val df = spark.createDataFrame(rowRDD,myschema)

3、方式三:直接读取一个具有格式的数据文件作为DataFrame(json文件)

val peopleDF = spark.read.json("/root/training/")

4、操作DataFrame:DSL语句和SQL语句

DSL语句:empDF.show

empDF.printSchema

查询所有员工的信息:df.show

查询所有员工的姓名:df.select("ename").show

或者df.select($"ename").show

查询员工信息:姓名 薪水 薪水+100

df.select($"ename",$"sal",$"sal"+100).show

查询工资大于2000的员工

df.filter("sal">2000).show

分组:

df.groupBy("deptno").count.show

SQL语句:需要将DataFrame注册成一张临时视图

df.createOrReplaceTempView("emp")

spark.sql("select * from emp").show

spark.sql("select * from emp where deptno=10").show

5、临时视图:2种

1、只在当前会话中有效:临时视图 df.createOrReplaceTempView("emp")

2、在全局范围内都有效:全局临时视图 df.createGlobalTempView("empG")

例:在当前会话中

spark.sql("select * from emp").show

spark.sql("select * from global_temp.empG").show

例:在新的会话中

spark.newSession.sal("select * from emp").show

spark.newSession.sal("select * from global_temp.empG").show

技术分享图片

二、使用数据源:

1、load函数加载数据源和save函数保存数据源

load函数默认的数据源是parquet文件

json函数默认的数据源是json文件

val usersDF = spark.read.load("/root/training/spakr-2.1.0-bin-hadoop2.7/examples/")

usersDF.select("name","favorite_color").show

usersDF.select("name","favorite_color").write.save("/root/temp/result")

2、Parquet文件:是sparkSQL load函数默认加载的数据源,按列存储的文件

如何把其他文件格式转换成parquet文件?

例:json文件---->parquet文件

val empJSON = spark.read.json("/root/temp/emp.json") #直接读取一个具有格式的数据文件作为DataFrame

empJSON.write.parquet("/root/temp/empparquet") #/empparquet目录不能事先存在

或者empJSON.wirte.mode("overwrite").parquet("/root/temp/result") #/result目录可以事先存在

功能:支持Schema的合并

第一个文件:val df1 = sc.makeRDD(1 to 5).map(i=>(i,i*2)).toDF("single","double")

df1.write.parquet("/root/temp/test_table/key=1")

第二个文件:val df2 = sc.makeRD(6 to 10).map(i=>(i,i*3)).toDF("single","triple")

df2.write.parquet("/root/temp/test_table/key=2")

合并两个文件:val df3 = spark.read.option("mergeSchema","true").parquet("/root/temp/test_table")

3、json文件:

spark.read.json("/root/training/spark-2.1.0-bin-hadoop-2.7/examples/src/main/resources/people.json")

spark.read.format("json").load("/root/training/spark-2.1.0-bin-hadoop2.7/examples/src/main/resources/people.json")

4、RDBMS:需要把RDBMS的驱动加入到spark shell中

spark.read.format("jdbc").option("url","jdbc:oracle:thin:@192.168.182.11:1521/orcl.example.com").option("dbtable","scott.emp").option("user","scott").option("password","tiger").load

或使用Properties类

import java.util.Properties

val prop = new Properties()

prop.setProperty("user","scott")

prop.setProperty("password","tiger")

val oracleDF1 = spark.read.jdbc("jdbc:oracle:thin:@192.168.182.11:1521/orcl")

作者:李金泽AllenLi,清华大学在读硕士,研究方向:大数据和人工智能

详细解读Spark的数据分析引擎:Spark SQL

标签:保存   overwrite   worker   ase   body   兼容   form   result   print   

原文地址:https://www.cnblogs.com/lijinze-tsinghua/p/8505281.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!