黑马程序员技术交流社区

标题: 【广州校区】Spark系列——从零学习SparkSQL编程(上) [打印本页]

作者: 白猫    时间: 2019-6-26 16:40
标题: 【广州校区】Spark系列——从零学习SparkSQL编程(上)
1.概述

早期的shark是spark团队在Hive之上设计的一套大规模数据仓库系统,shark继承了hive的复杂代码,并依赖spark
的版本,维护成本高,2014年官方从shark转战Spark SQL.spark SQL底层使用的是RDD 效率非常高。
SparkSQL的优点如下:
1. sql查询与spark rdd无缝整合。
2. 可以使用java,scala,python,R语言等来实现。
3. 可以对接多种数据源:Hive Avro ORC JSON JDBC/ODBC 等,同时兼容Hive语法。

2.核心API

SparkSQL的核心是DataFrame,在Spark 2.0以后,DataFrame和DataSet已经整合在一起。这里我们重点讲解
DataFrame:
1. DataFrame分为两部分:1.RDD基础分布式数据集。 2.Schema元信息(记录二维数据表中每一列的列名和类型)
2. 不同于RDD记录的一个个对象(spark不知道对象内部的信息),DataFrame却提供了每条数据的内部多列详细
信息。
3. DataFrame SQL的代码会经过Catalyst(Spark优化器)的优化,变成高效的处理代码。举个例子吧,比如使用
RDD读取某个文件,并取出第1条数据,此刻系统会先读取所有数据,再取第一条,消费了大部分没用的内存,
而SparkSQL经过优化后,会只将第一条数据加载到内存中。

3.利用spark-shell创建DataFrame


[root@node01 ~]# spark-shell --master spark://node01:7077Spark context available as 'sc' (master = spark://node01:7077, app id = app-20190619142652-0000).
Spark session available as 'spark'.

1. sc:SparkContext是所有Spark任务的入口,它包含了Spark程序的基本设置,如程序名字 内存大小 并行处理
粒度。
2. spark session:不仅有SparkContext的所有功能,还集成了DataFrame,Spark Streaming,Structed Streaming
提供的API,不用为不同功能分别定义Context。
要操作DataFrame,首先要知道如何创建,这介绍两种方式:
方式一:将RDD转换成DataFrame,这里case class就是定义DataFrame的schema

// 将spark/examples/src/main/resource下有各种资源,存放到hdfs上。
scala> val rdd1 = sc.textFile("/spark_res/people.txt").map(_.split(", "))
scala> case class People(val name:String,val age:Int)
scala> val rdd2 = rdd1.map(x => People(x(0),x(1).toInt))
scala> rdd2.toDF
这里报错:已存在元数据,但对本案例无影响
res3: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> res3.printSchema 查看数据结构
scala> res3.show    查看数据内容


方式二:直接通过spark session读取【普通文本数据、JSON数据、Parquet数据 】
scala> val df = spark.read.text("/spark_res/people.txt")df: org.apache.spark.sql.DataFrame = [value: string]
scala> val rdd1 = spark.read.json("/spark_res/people.json")
rdd1: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
scala> val rdd1 = spark.read.parquet("/spark_res/users.parquet")
rdd1: org.apache.spark.sql.DataFrame = [name: string, favorite_color: string ... 1 more
field]


方式三:直接通过spark session读取 【Mysql数据】
1.启动spark-shell指定驱动包
spark-shell \
--master spark://node01:7077 \
--executor-memory 1g \
--total-executor-cores 2 \
--jars /export/servers/hive-1.1.0-cdh5.14.0/lib/mysql-connector-java-5.1.38.jar \--driver-class-path /export/servers/hive-1.1.0-cdh5.14.0/lib/mysql-connector-java-5.1.38.jar

2.从mysql中加载数据
val mysqlDF = spark.read.format("jdbc").options(
Map("url" -> "jdbc:mysql://192.168.52.105:3306/iplocation",
"driver" -> "com.mysql.jdbc.Driver","dbtable" -> "iplocation",
"user" -> "root", "password" -> "123456")).load()


scala> mysqlDF.show
+----------+---------+-----------+
| longitude| latitude|total_count|
+----------+---------+-----------+
|108.948024|34.263161|    1824|
|116.405285|39.904989|    1535|
| 107.7601| 29.32548|     85|
+----------+---------+-----------+




4.利用spark-shell 操作DataFrame
上一节我们介绍了如何创建DataFrame,接着看看如何使用DataFrame操作数据,对于操作数据,系统提供了两种
方式:
操作方式一:DSL风格语法
scala> df1.select("name").show
scala> df1.select($"age",$"age"+1).show
scala> df1.filter("age is not null").select($"age",$"age"+1).show
scala> df1.filter(col("age")>21).show

操作方式二:SQL语法
scala> val df = rdd2.toDF
scala> val df = rdd2.toDF
scala> spark.sql("select * from t_people").show
scala> spark.sql("select * from t_people where age>20").show



下一节,我们将介绍如何使用Java代码来操作 DataFrame。







欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/) 黑马程序员IT技术论坛 X3.2