Spark入门

第5章 Spark入门
Spark是加州大学伯克利分校AMP实验室(Algorithms, Machines, and People Lab)开发通用内存并行计算框架。Spark在2013年6月进入Apache成为孵化项目,8个月后成为Apache顶级项目,2014年至2015年,Spark经历了高速发展,Databricks 2015 Spark调查报告显示:2014年9月至2015年9月,已经有超过600个Spark源码贡献者,而在此之前的12个月人数只有315,Spark超越Hadoop,无可争议地成为大数据领域内最活跃的开源项目。除此之外,已经有超过200个公司为Spark奉献过源代码,使Spark社区成为迄今为止开发人员参与最多的社区。
Spark以其先进的设计理念,迅速成为社区的热门项目,围绕着Spark推出了Spark SQL、Spark Streaming、MLLib和GraphX等组件,也就是BDAS(伯克利数据分析栈),这些组件逐渐形成大数据处理一站式解决平台。
Spark使用Scala语言进行实现,它是一种面向对象、函数式编程语言,能够像操作本地集合对象一样轻松地操作分布式数据集(Scala 提供一个称为 Actor 的并行模型,其中Actor通过它的收件箱来发送和接收非同步信息而不是共享数据,该方式被称为:Shared Nothing 模型)。在Spark官网上介绍,它具有运行速度快、易用性好、通用性强和随处运行等特点。
 运行速度快
Spark拥有DAG执行引擎,支持在内存中对数据进行迭代计算。官方提供的数据表明,如果数据由磁盘读取,速度是Hadoop MapReduce的10倍以上,如果数据从内存中读取,速度可以高达100多倍。

 易用性好
Spark不仅支持Scala编写应用程序,而且支持Java和Python等语言进行编写,特别是Scala是一种高效、可拓展的语言,能够用简洁的代码处理较为复杂的处理工作。
 通用性强
Spark生态圈包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等组件,这些组件分别处理Spark Core提供内存计算框架、SparkStreaming的实时处理应用、Spark SQL的即席查询、MLlib或MLbase的机器学习和GraphX的图处理,它们无缝集成于Spark平台。
 随处运行
Spark具有很强的适应性,能够读取HDFS、HBase、S3和Techyon等的数据,能够以Mesos、YARN和自身携带的Standalone作为资源管理器调度job,来完成Spark应用程序的计算。

5.1 Spark架构及原理
进入交换式编程界面,在之前的Spark版本中,Spark shell会自动创建一个SparkContext对象sc。2.0中Spark shell则会自动创建一个SparkSession对象(spark),在输入spark时就会发现它已经存在了。下图说明了SparkContext在Spark中的主要功能。

从图中可以看到SparkContext起到的是一个中介的作用,通过它来使用Spark其他的功能。每一个JVM都有一个对应的SparkContext,Driver program通过SparkContext连接到集群管理器来实现对集群中任务的控制。Spark配置参数的设置以及对SQLContext、HiveContext和StreamingContext的控制也要通过SparkContext。不过在Spark2.0中上述的一切功能都是通过SparkSession来完成的,同时SparkSession也简化了DataFrame/Dataset API的使用和对数据的操作。SparkSession成为Spark的一个全新的切入点。

5.2 Spark SQL简介
Spark Sql 的前身是shark,最初是用来查询Hive,Hive是为熟悉数据库,但不熟悉MapReduce的开发人员提供的工具,Hive提供可以对数据进行提取转化和加载(简称ETL)功能,通过Hive工具可以查询分析存储在Hadoop上的大规模数据。Hive定义了简单的查询语言HQL(类SQL),可以把SQL操作转成MapReduce任务。
MapReduce的计算过程消耗大量的IO,降低了运行效率,为了提高查询Hadoop上的数据的效率,出现了很多工具,如Impla 和shark等,其中Shark是spark的生态组件之一,它复用了Hive的SQL解析等组件,修改了内存管理,物理计划、执行模块,使它能够运行在Spark的计算引擎上,使用SQL的查询速度有了10-100倍的提升。
随着shark的发展,shark对Hive的依赖限制了其发展,包括语法解析器和查询优化器。Spark团队汲取了shark的优点重新设计了Spark SQL,使之在数据兼容、性能优化、组件扩展等方面得到极大的提升。
 数据兼容:不仅兼容Hive,还可以从RDD、parquet文件、Json文件获取数据、支持从RDBMS获取数据。
 性能优化:采用内存列式存储、自定义序列化器等方式提升性能。
 组件扩展:SQL的语法解析器、分析器、优化器都可以重新定义和扩展。
Spark SQL是Spark提供的针对结构化数据处理的模块。不同于基本的Spark RDD API,SparkSQL提供的接口提供了更多的关于数据和计算执行的信息。其数据源可以是文件、Hive表、其他数据库等,我们可以通过command-line或者JDBC/ODBC与SQL接口进行交互。

5.3 Spark SQL操作MySQL

//启动spark客户端
spark-shell --master spark://master:7077 --driver-memory 1G
//连接mysql数据库
scala>val jdbcDF = spark.read.format("jdbc").options(Map("url" ->"jdbc:mysql://192.168.1.106:3306/testdb", "driver" ->"com.mysql.jdbc.Driver", "dbtable" ->"testdb.stud_info", "user" ->"feigu", "password" ->"feigu")).load()
//创建一个视图
scala>jdbcDF.createOrReplaceTempView("stud_info")
scala> val sqlDF=sql("select * from stud_info limit 5")
sqlDF: org.apache.spark.sql.DataFrame = [stud_code: string, stud_name: string ... 8 more fields]

scala> sqlDF.show()
+----------+---------+--------+------+--------+------------+------------+-----+
| stud_code|stud_name|stud_sex| birthday| log_date|orig_addr|lev_date|college_code|college_name|state|
+----------+---------+--------+----------+---+------------+------------+-----+
|2015101000| 王进| M|1997-08-01|2014-09-01| 苏州| null| 10| 理学院| 1
|2015101001| 刘海| M|1997-09-29|2014-09-01| 上海| null| 10| 理学院| 1
|2015101002| 张飞| M|1996-10-21|2014-09-02| 济南| null| 10| 理学院| 1
|2015101003| 刘婷| F|1998-01-10|2014-09-01| 北京| null| 10| 理学院| 1
|2015101004| 卢家| M|1997-08-01|2014-09-01| 南京| null| 10| 理学院| 1

5.4 Spark SQL操作Hive
Spark SQL与Hive的交换,当在Hive上工作时,必须实例化SparkSession对Hive的支持,包括对持久化Hive元存储的连通性,对Hive序列化反序列化,Hive用户自定义函数的支持。当没有在hive-site.xml配置是,context会自动在当前目录创建metastore_db并且创建一个被spark.sql.warehouse.dir配置的目录,默认在spark应用启动的当前目录的spark-warehouse。注意从Spark2.0.0开始hive-site.xml中的hive.metastore.warehouse.dir参数被弃用。作为替代,使用spark.sql.warehouse.dir来指定仓库中数据库的位置。你可能需要授权写权限给启动spark应用的用户。

spark-shell --master spark://master:7077 --driver-memory 1G --total-executor-cores 2
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

case class Record(key: Int, value: String)

// warehouseLocation points to the default location for managed databases and tables
val warehouseLocation = "spark-warehouse"

val spark = SparkSession
.builder()
.appName("Spark Hive Example")
.config("spark.sql.warehouse.dir", warehouseLocation)
.enableHiveSupport()
.getOrCreate()

import spark.implicits._
import spark.sql

sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)")
sql("LOAD DATA LOCAL INPATH '/home/hadoop/bigdata/spark/examples/src/main/resources/kv1.txt' INTO TABLE src")

// Queries are expressed in HiveQL
sql("SELECT * FROM src limit 3").show()

// +---+-------+
// |key| value|
// +---+-------+
// |238|val_238|
// | 86| val_86|
// |311|val_311|
// ...

// Aggregation queries are also supported.
sql("SELECT COUNT(*) FROM src").show()
// +--------+
// |count(1)|
// +--------+
// | 500 |
// +--------+

// The results of SQL queries are themselves DataFrames and support all normal functions.
val sqlDF = sql("SELECT key, value FROM src WHERE key < 10 ORDER BY key") // The items in DaraFrames are of type Row, which allows you to access each column by ordinal. val stringsDS = sqlDF.map { case Row(key: Int, value: String) => s"Key: $key, Value: $value"
}
stringsDS.show()
// +--------------------+
// | value|
// +--------------------+
// |Key: 0, Value: val_0|
// |Key: 0, Value: val_0|
// |Key: 0, Value: val_0|
// ...

// You can also use DataFrames to create temporary views within a SparkSession.
val recordsDF = spark.createDataFrame((1 to 100).map(i => Record(i, s"val_$i")))
recordsDF.createOrReplaceTempView("records")

// Queries can then join DataFrame data with data stored in Hive.
sql("SELECT * FROM records r JOIN src s ON r.key = s.key").show()
// +---+------+---+------+
// |key| value|key| value|
// +---+------+---+------+
// | 2| val_2| 2| val_2|
// | 4| val_4| 4| val_4|
// | 5| val_5| 5| val_5|

5.5 pyspark操作hive
以下代码操作hive中表:fund.rpt_fund的数据,fund是数据库名称,rpt_fund是表名。

from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.sql.functions import *

warehouse_location = 'spark-warehouse'

spark = SparkSession \
.builder \
.appName("Python Spark SQL Hive integration example") \
.config("spark.sql.warehouse.dir", warehouse_location) \
.enableHiveSupport() \
.getOrCreate()

df= spark.sql("Select fund_date,nav,accnav from fund.rpt_fund")
df1=df.select(substring(trim(df.fund_date),1,7).alias('year'),df.nav,df.accnav)
df2=df1.groupBy("year").mean("nav","accnav").orderBy("year")
df3=df2.toPandas()
df4=df3.set_index(['year'])
df4.plot(kind='bar',rot=30)
## df4.plot(kind='line',rot=30)

或折线图

5.6练习
一、单选题
1. Spark 的四大组件下面哪个不是
A.Spark Streaming
B Mlib
C Graphx
D Spark R

2.spark 1.4 版本的最大变化
A spark sql Release 版本
B 引入 Spark R
C DataFrame
D支持动态资源分配

3. Spark Job 默认的调度模式
A FIFO B FAIR
C 无 D 运行时指定

4.下面哪个不是 RDD 的特点
A. 可分区 B 可序列化 C 可修改 D 可持久化

5. 关于广播变量,下面哪个是错误的
A 任何函数调用 B 是只读的 C 存储在各个节点 D 存储在磁盘或 HDFS

6. 关于累加器,下面哪个是错误的
A 支持加法 B 支持数值类型
C 可并行 D 不支持自定义类型

7.Spark 支持的分布式部署方式中哪个是错误的
A standalone B spark on mesos
C spark on YARN D Spark on local

8.Stage 的 Task 的数量由什么决定
A Partition B Job C Stage D TaskScheduler

9.下面哪个操作是窄依赖
A join B filter
C group D sort

10.下面哪个操作肯定是宽依赖
A map B flatMap
C reduceByKey D sample

11.spark 的 master 和 worker 通过什么方式进行通信的?
A http B nio C netty D Akka

12.下列哪个不是 RDD 的缓存方法
A persist() B Cache()
C Memory()

13.Task 运行在下来哪里个选项中 Executor 上的工作单元
A Driver program B. spark master
C.worker node D Cluster manager

14.hive 的元数据存储在 derby 和 MySQL 中有什么区别
A.没区别 B.多会话 C.支持网络环境 D数据库的区别

15.DataFrame 和 RDD 最大的区别
A.科学统计支持 B.多了 schema
C.存储方式不一样 D.外部数据源支持

二、实战题
1、利用Spark SQL查询MySQL数据库中stud_score,stud_info表,统计学生各科总成绩,并选出前5名的同学。

附录;推荐读物或网站
一、Linux
1、网站:
Linux中国:https://linux.cn
鸟哥的Linux私房菜: http://linux.vbird.org
Linux下载站: http://www.linuxdown.net
Linux公社: http://www.linuxidc.com
2、图书:
《鸟哥的Linux基础学习篇》鸟哥著

二、MySQL
1、网站:
MySQL社区:http://www.mysqlpub.com/
MySQL菜鸟教程: http://www.runoob.com/mysql/mysql-tutorial.html
MySQL官网:http://www.mysql.com/
2、图书
《深入浅出MySQL》唐汉名等著
三、Python
1、网站:
http://www.pythondoc.com/
2、图书:
基础入门:《python编程入门》Toby Donaldson著
《python基础教程》Magnus Lie Hetland著
数据分析:《利用Python进行数据分析》
机器学习:《机器学习实战》
《Python数据挖掘入门与实践》
《Python机器学习》【美】sebastian Raschka
四、Hadoop
1、hadoop官网:
http://www.apache.org/
2、图书:
《hadoop权威指南》第三版
《Hadoop大数据处理》刘军著
《自己动手做大数据系统》飞谷团队著
五、Spark
1、Spark官网:
http://www.apache.org/
2、图书
《Spark 核心技术与高级应用》于俊等著
《spark机器学习》【南非】Nick Pentreath 著
六.R
1.网站:
http://ggplot2.tidyverse.org/reference/
2. 图书:
《R语言实战》作者: Robert I. Kabacoff 译者: 高涛 / 肖楠 / 陈钢
《ggplot2数据分析与图形艺术》[美] 哈德利•威克姆 著;统计之都 译

 

发表评论