Spark-Streaming进阶与Spark优化

时间:2021-10-21 00:41:34

Spark课堂笔记

Spark生态圈:
Spark Core : RDD(弹性分布式数据集)
Spark SQL
Spark Streaming

Spark MLLib:协同过滤,ALS,逻辑回归等等 --> 机器学习
Spark Graphx : 图计算

重点在前三章

-----------------Spark Core------------------------
一、什么是Spark?特点?
https://spark.apache.org/
Apache Spark™ is a unified analytics engine for large-scale data processing.

特点:快、易用、通用性、兼容性(完全兼容Hadoop)

快:快100倍(Hadoop 3 之前)
易用:支持多种语言开发
通用性:生态系统全。
易用性:兼容Hadoop

spark 取代 Hadoop

二、安装和部署Spark、Spark 的 HA

1、spark体系结构
Spark的运行方式

Yarn

Standalone:本机调试(demo)

Worker:从节点。每个服务器上,资源和任务的管理者。只负责管理一个节点。

执行过程:
一个Worker 有多个 Executor。 Executor是任务的执行者,按阶段(stage)划分任务。————> RDD

客户端:Driver Program 提交任务到集群中。

1、spark-submit
2、spark-shell

2、spark的搭建
(1)准备工作:JDK 配置主机名 免密码登录
(2)伪分布式模式
在一台虚拟机上模拟分布式环境(Master和Worker在一个节点上)

export JAVA_HOME=/usr/java/jdk1.8.0_201
export SPARK_MASTER_HOST=node3
export SPARK_MASTER_PORT=7077

(3)全分布式环境
修改slave文件 拷贝到其他两台服务器 启动

3、Spark的 HA
回顾HA;
(*)HDFS Yarn Hbase Spark 主从结构
(*)单点故障

(1)基于文件目录的单点恢复
(*)本质:还是只有一个主节点Master,创建了一个恢复目录,保存集群状态和任务的信息。
当Master挂掉,重新启动时,会从恢复目录下读取状态信息,恢复出来原来的状态

用途:用于开发和测试,生产用zookeeper
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=FILESYSTEM
-Dspark.deploy.recoveryDirectory=/usr/local/spark-2.1.0-bin-hadoop2.7/recovery"

(2)基于Zookeeper :和Hadoop类似

(*)复习一下zookeeper:
相当于一个数据库,把一些信息存放在zookeeper中,比如集群的信息。
数据同步功能,选举功能,分布式锁功能

数据同步:给一个节点中写入数据,可以同步到其他节点

选举:Zookeeper中存在不同的角色,Leader Follower。如果Leader挂掉,重新选举Leader

分布式锁:秒杀。以目录节点的方式来保存数据。

修改 spark-env.sh
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=node3:2181,node4:2181,node5:2181
-Dspark.deploy.zookeeper.dir=/spark"

同步到其他两台服务器。

在node3 start-all node3 master node4 Worker node5 Worker
在node4 start-master node3 master node4 master(standby) node4 Worker node5 Worker

在node3上kill master
node4 master(Active) node4 Worker node5 Worker

在网页http://192.168.109.134:8080/ 可以看到相应信息

三、执行Spark的任务:两个工具

1、spark-submit:用于提交Spark的任务
任务:jar。

举例:蒙特卡洛求PI(圆周率)。

./spark-submit --master spark://node3:7077 --class

--class指明主程序的名字

/usr/local/spark-2.1.0-bin-hadoop2.7/bin/spark-submit --master spark://node3:7077
--class org.apache.spark.examples.SparkPi
/usr/local/spark-2.1.0-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.0.jar 100

2、spark-shell 相当于REPL
作为一个独立的Application运行

两种模式:
(1)本地模式
spark-shell 后面不接任何参数,代表本地模式

Spark context available as 'sc' (master = local[*], app id = local-1554038459298).

sc 是 SparkContext 对象名。 local[*] 代表本地模式,不提交到集群中运行。

(2)集群模式
./spark-submit --master spark://node3:7077 提交到集群中运行

Spark context available as 'sc' (master = spark://node3:7077, app id = app-20190331212447-0000).

master = spark://node3:7077

Spark session available as 'spark'
Spark Session 是 2.0 以后提供的,利用 SparkSession 可以访问spark所有组件。

示例:WordCount程序

(*)处理本地文件,把结果打印到屏幕上
scala> sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.collect

res0: Array[(String, Int)] = Array((is,1), (love,2), (capital,1), (Beijing,2), (China,2), (I,2), (of,1), (the,1))

(*)处理HDFS文件,结果保存在hdfs上
sc.textFile("hdfs://node1:8020/tmp_files/test_WordCount.txt")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.saveAsTextFile("hdfs://node1:8020/output/0331/test_WordCount")

-rw-r--r-- 3 root supergroup 0 2019-03-31 21:43 /output/0331/test_WordCount/_SUCCESS
-rw-r--r-- 3 root supergroup 40 2019-03-31 21:43 /output/0331/test_WordCount/part-00000
-rw-r--r-- 3 root supergroup 31 2019-03-31 21:43 /output/0331/test_WordCount/part-00001

_SUCCESS 代表程序执行成功

part-00000 part-00001 结果文件,分区。里面内容不重复。

(*)单步运行WordCount ----> RDD

scala> val rdd1 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
rdd1: org.apache.spark.rdd.RDD[String] = /usr/local/tmp_files/test_WordCount.txt MapPartitionsRDD[12] at textFile at <console>:24

scala> 1+1
res2: Int = 2

scala> rdd1.collect
res3: Array[String] = Array(I love Beijing, I love China, Beijing is the capital of China)

scala> val rdd2 = rdd1.flatMap(_.split(" "))
rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at flatMap at <console>:26

scala> rdd2.collect
res4: Array[String] = Array(I, love, Beijing, I, love, China, Beijing, is, the, capital, of, China)

scala> val rdd3 = rdd2.map((_,1))
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[14] at map at <console>:28

scala> rdd3.collect
res5: Array[(String, Int)] = Array((I,1), (love,1), (Beijing,1), (I,1), (love,1), (China,1), (Beijing,1), (is,1), (the,1), (capital,1), (of,1), (China,1))

scala> val rdd4 = rdd3.reduceByKey(_+_)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[15] at reduceByKey at <console>:30

scala> rdd4.collect
res6: Array[(String, Int)] = Array((is,1), (love,2), (capital,1), (Beijing,2), (China,2), (I,2), (of,1), (the,1))

RDD 弹性分布式数据集
(1)依赖关系 : 宽依赖和窄依赖
(2)算子:
函数:
Transformation : 延时计算 map flatMap textFile
Action : 立即触发计算 collect




说明:scala复习
(*)flatten:把嵌套的结果展开
scala> List(List(2,4,6,8,10),List(1,3,5,7,9)).flatten
res21: List[Int] = List(2, 4, 6, 8, 10, 1, 3, 5, 7, 9)


(*)flatmap : 相当于一个 map + flatten

scala> var myList = List(List(2,4,6,8,10),List(1,3,5,7,9))
myList: List[List[Int]] = List(List(2, 4, 6, 8, 10), List(1, 3, 5, 7, 9))

scala> myList.flatMap(x=>x.map(_*2))
res22: List[Int] = List(4, 8, 12, 16, 20, 2, 6, 10, 14, 18)

myList.flatMap(x=>x.map(_*2))

执行过程:
1、将 List(2, 4, 6, 8, 10), List(1, 3, 5, 7, 9) 调用 map(_*2) 方法。x 代表一个List
2、flatten
3、在IDE中开发scala版本和Java版本的WorkCount。

(1)scala版本的WordCount

新建一个工程,把jar引入到工程中。

export jar 点击下一步下一步,不需要设置main class

把jar上传到服务器上。

spark-submit --master spark://node3:7077
--class day1025.MyWordCount
/usr/local/tmp_files/Demo1.jar
hdfs://node2:8020/tmp_files/test_WordCount.txt
hdfs://node2:8020/output/1025/demo1

(2)java版本的WordCount

./spark-submit --master spark://node3:7077 --class day0330.JavaWordCount /usr/local/tmp_files/Demo2.jar

四、分析Spark的任务流程

1、分析WordCount程序处理过程
见图片

2、Spark调度任务的过程

提交到及群众运行任务时,spark执行任务调度。

见图片

五、RDD和RDD特性、RDD的算子

1、RDD:弹性分布式数据集
(*)Spark中最基本的数据抽象。
(*)RDD的特性
* Internally, each RDD is characterized by five main properties:
*
* - A list of partitions

1、是一组分区。
RDD由分区组成,每个分区运行在不同的Worker上,通过这种方式来实现分布式计算。

* - A function for computing each split
在RDD中,提供算子处理每个分区中的数据

* - A list of dependencies on other RDDs

RDD存在依赖关系:宽依赖和窄依赖。

* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)

可以自定义分区规则来创建RDD

* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
* an HDFS file)

优先选择离文件位置近的节点来执行


如何创建RDD?

(1)通过SparkContext.parallelize方法来创建
scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[32] at parallelize at <console>:29

scala> rdd1.partitions.length
res35: Int = 3

scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[33] at parallelize at <console>:29

scala> rdd1.partitions.length
res36: Int = 2

(2)通过外部数据源来创建
sc.textFile()

scala> val rdd2 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
rdd2: org.apache.spark.rdd.RDD[String] = /usr/local/tmp_files/test_WordCount.txt MapPartitionsRDD[35] at textFile at <console>:29


2、 算子
(1)Transformation

map(func):相当于for循环,返回一个新的RDD

filter(func):过滤
flatMap(func):flat+map 压平


mapPartitions(func):对RDD中的每个分区进行操作
mapPartitionsWithIndex(func):对RDD中的每个分区进行操作,可以取到分区号。

sample(withReplacement, fraction, seed):采样

集合运算
union(otherDataset)
intersection(otherDataset)

distinct([numTasks])):去重

聚合操作:group by
groupByKey([numTasks])
reduceByKey(func, [numTasks])
aggregateByKey(zeroValue)(seqOp,combOp,[numTasks])

排序
sortByKey([ascending], [numTasks])
sortBy(func,[ascending], [numTasks])

join(otherDataset, [numTasks])
cogroup(otherDataset, [numTasks])
cartesian(otherDataset)
pipe(command, [envVars])
coalesce(numPartitions)

重分区:
repartition(numPartitions)
repartitionAndSortWithinPartitions(partitioner)

举例:
1、创建一个RDD,每个元素乘以2,再排序
scala> val rdd1 = sc.parallelize(Array(3,4,5,100,79,81,6,8))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[36] at parallelize at <console>:29

scala> val rdd2 = rdd1.map(_*2)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[37] at map at <console>:31

scala> rdd2.collect
res37: Array[Int] = Array(6, 8, 10, 200, 158, 162, 12, 16)

scala> rdd2.sortBy(x=>x,true).collect
res39: Array[Int] = Array(6, 8, 10, 12, 16, 158, 162, 200)

scala> rdd2.sortBy(x=>x,false).collect
res40: Array[Int] = Array(200, 162, 158, 16, 12, 10, 8, 6)

def sortBy[K](f: (T) ⇒ K, ascending: Boolean = true)

过滤出大于20的元素:

scala> val rdd3 = rdd2.filter(_>20)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[53] at filter at <console>:33

scala> rdd3.collect
res41: Array[Int] = Array(200, 158, 162)

2、字符串(字符)类型的RDD

scala> val rdd4 = sc.parallelize(Array("a b c","d e f","g h i"))
rdd4: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[54] at parallelize at <console>:29

scala> rdd4.flatMap(_.split(" ")).collect
res42: Array[String] = Array(a, b, c, d, e, f, g, h, i)

3、RDD的集合运算:

scala> val rdd6 = sc.parallelize(List(1,2,3,6,7,8,9,100))
rdd6: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[56] at parallelize at <console>:29

scala> val rdd7 = sc.parallelize(List(1,2,3,4))
rdd7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[57] at parallelize at <console>:29

scala> val rdd8 = rdd6.union(rdd7)
rdd8: org.apache.spark.rdd.RDD[Int] = UnionRDD[58] at union at <console>:33

scala> rdd8.collect
res43: Array[Int] = Array(1, 2, 3, 6, 7, 8, 9, 100, 1, 2, 3, 4)

scala> rdd8.distinct.collect
res44: Array[Int] = Array(100, 4, 8, 1, 9, 6, 2, 3, 7)


4、分组操作:reduceByKey

<key value>
scala> val rdd1 = sc.parallelize(List(("Tom",1000),("Andy",2000),("Lily",1500)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[62] at parallelize at <console>:29

scala> val rdd2 = sc.parallelize(List(("Andy",1000),("Tom",2000),("Mike",500)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[63] at parallelize at <console>:29

scala> val rdd3 = rdd1 union rdd2
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[64] at union at <console>:33

scala> rdd3.collect
res45: Array[(String, Int)] = Array((Tom,1000), (Andy,2000), (Lily,1500), (Andy,1000), (Tom,2000), (Mike,500))

scala> val rdd4= rdd3.groupByKey
rdd4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[65] at groupByKey at <console>:35

scala> rdd4.collect
res46: Array[(String, Iterable[Int])] = Array(
(Tom,CompactBuffer(1000, 2000)),
(Andy,CompactBuffer(2000, 1000)),
(Mike,CompactBuffer(500)), (
Lily,CompactBuffer(1500)))

scala> rdd3.reduceByKey(_+_).collect
res47: Array[(String, Int)] = Array((Tom,3000), (Andy,3000), (Mike,500), (Lily,1500))

reduceByKey will provide much better performance.

官方不推荐使用 groupByKey 推荐使用 reduceByKey

5、cogroup

scala> val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[67] at parallelize at <console>:29

scala> val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[68] at parallelize at <console>:29

scala> val rdd3 = rdd1.cogroup(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, (Iterable[Int], Iterable[Int]))] = MapPartitionsRDD[70] at cogroup at <console>:33

scala> rdd3.collect
res48: Array[(String, (Iterable[Int], Iterable[Int]))] = Array(
(tom,(CompactBuffer(1, 2),CompactBuffer(1))),
(jerry,(CompactBuffer(3),CompactBuffer(2))),
(shuke,(CompactBuffer(),CompactBuffer(2))),
(kitty,(CompactBuffer(2),CompactBuffer())))

6、reduce操作(Action)

聚合操作
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at parallelize at <console>:29

scala> rdd1.reduce(_+_)
res49: Int = 15


7、需求:按照value排序。
做法:
1、交换,把key 和 value交换,然后调用sortByKey方法
2、再次交换

scala> val rdd1 = sc.parallelize(List(("tom",1),("jerry",3),("ketty",2),("shuke",2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[72] at parallelize at <console>:29

scala> val rdd2 = sc.parallelize(List(("jerry",1),("tom",3),("shuke",5),("ketty",1)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[73] at parallelize at <console>:29

scala> val rdd3 = rdd1.union(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[74] at union at <console>:33

scala> val rdd4 = rdd3.reduceByKey(_+_)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[75] at reduceByKey at <console>:35

scala> rdd4.collect
res50: Array[(String, Int)] = Array((tom,4), (jerry,4), (shuke,7), (ketty,3))

scala> val rdd5 = rdd4.map(t=>(t._2,t._1)).sortByKey(false).map(t=>(t._2,t._1))
rdd5: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[80] at map at <console>:37

scala> rdd5.collect
res51: Array[(String, Int)] = Array((shuke,7), (tom,4), (jerry,4), (ketty,3))


(2)Action

reduce(func)

collect()
count()
first()
take(n)
takeSample(withReplacement,num, [seed])
takeOrdered(n, [ordering])
saveAsTextFile(path)
saveAsSequenceFile(path) 
saveAsObjectFile(path) 
countByKey()

foreach(func):与map类似,没有返回值。

3、特性:
(1)RDD的缓存机制
(*)作用:提高性能
(*)使用:标识RDD可以被缓存 persist cache

(*)可以缓存的位置:
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

/**
* Persist this RDD with the default storage level (`MEMORY_ONLY`).
*/
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

/**
* Persist this RDD with the default storage level (`MEMORY_ONLY`).
*/
def cache(): this.type = persist()


举例:测试数据,92万条

scala> val rdd1 = sc.textFile("hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[82] at textFile at <console>:29

scala> rdd1.count --> 直接出发计算
res52: Long = 923452

scala> rdd1.cache --> 标识RDD可以被缓存,不会触发计算
res53: rdd1.type = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[82] at textFile at <console>:29

scala> rdd1.count --> 和第一步一样,触发计算,但是,把结果进行缓存
res54: Long = 923452

scala> rdd1.count --> 从缓存中直接读出结果
res55: Long = 923452

(2)RDD的容错机制:通过检查点来实现。

/**
* Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint
* directory set with `SparkContext#setCheckpointDir` and all references to its parent
* RDDs will be removed. This function must be called before any job has been
* executed on this RDD. It is strongly recommended that this RDD is persisted in
* memory, otherwise saving it on a file will require recomputation.
*/
(*)复习检查点:HDFS中的检查点:有SecondaryNamenode来实现日志的合并。

(*)RDD的检查点:容错
概念:血统 Lineage
理解:表示任务执行的生命周期。
WordCount textFile ---> redceByKey

如果血统越长,越容易出错。

假如有检查点,可以从最近的一个检查点开始,往后面计算。不用重头计算。

(*)RDD检查点的类型:

(1)基于本地目录:需要将Spark shell 或者任务运行在本地模式上(setMaster("local"))

开发和测试

(2)HDFS目录:用于生产。

sc.setCheckPointDir(目录)

举例:设置检查点
scala> var rdd1 = sc.textFile("hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[1] at textFile at <console>:24

设置检查点目录:
scala> sc.setCheckpointDir("hdfs://192.168.109.131:8020/sparkckpt")

标识rdd1可以执行检查点操作
scala> rdd1.checkpoint

scala> rdd1.count
res2: Long = 923452

(3)依赖关系:宽依赖,窄依赖。
划分任务执行的stage

见讲义。

六、RDD的高级算子
1、mapPartitionsWithIndex:对RDD中的每个分区(带有下标)进行操作,下标用index表示
通过这个算子,我们可以获取分区号。

def mapPartitionsWithIndex[U](
f: (Int, Iterator[T]) ⇒ Iterator[U],
preservesPartitioning: Boolean = false)(
implicit arg0: ClassTag[U]): RDD[U]

参数:f是个函数参数 f 中第一个参数是Int,代表分区号,第二个Iterator[T]代表分区中的元素

举例:把分区中的元素,包括分区号,都打印出来。

scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at <console>:24

scala> def fun1(index:Int, iter:Iterator[Int]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }
fun1: (index: Int, iter: Iterator[Int])Iterator[String]

scala> rdd1.mapPartitions
mapPartitions mapPartitionsWithIndex

scala> rdd1.mapPartitionsWithIndex(fun1).collect
res3: Array[String] = Array(
[partId : 0 , value = 1 ], [partId : 0 , value = 2 ],
[partId : 1 , value = 3 ], [partId : 1 , value = 4 ], [partId : 1 , value = 5 ],
[partId : 2 , value = 6 ], [partId : 2 , value = 7 ], [partId : 2 , value = 8 ])

2、aggregate:聚合操作。类似于分组。

(*)先对局部进行聚合操作,再对全局进行聚合操作。

调用聚合操作
scala> val rdd2 = sc.parallelize(List(1,2,3,4,5),2)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24

scala> rdd2.mapPartitionsWithIndex(fun1).collect
res4: Array[String] = Array(
[partId : 0 , value = 1 ], [partId : 0 , value = 2 ],
[partId : 1 , value = 3 ], [partId : 1 , value = 4 ], [partId : 1 , value = 5 ])

scala> import scala.math._
import scala.math._

scala> rdd2.aggregate(0)(max(_,_),_+_)
res6: Int = 7

说明:aggregate
(0) 初始值是 0
(max(_,_) 局部操作的函数
, _+_ 全局操作的函数
)

scala> rdd2.aggregate(100)(max(_,_),_+_)
res8: Int = 300

分析结果:初始值是100,代表每个分区多了一个100
全局操作,也多了一个100
100+100+100 = 300


对RDD中的元素进行求和
1、RDD.map
2、聚合操作

scala> rdd2.aggregate(0)(_+_,_+_)
res9: Int = 15

MapReduce Combiner

scala> rdd2.aggregate(10)(_+_,_+_)
res10: Int = 45

(*)对字符串操作
scala> val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[7] at parallelize at <console>:27

scala> rdd2.aggregate("")(_+_,_+_)
res11: String = abcdef

scala> rdd2.aggregate("*")(_+_,_+_)
res12: String = **def*abc

结果分析:
1、*abc *def
2、**def*abc

(*)复杂的例子:
1、
scala> val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[8] at parallelize at <console>:27

scala> def fun1(index:Int, iter:Iterator[String]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }

scala> rdd3.mapPartitionsWithIndex(fun1).collect
res17: Array[String] = Array(
[partId : 0 , value = 12 ], [partId : 0 , value = 23 ],
[partId : 1 , value = 345 ], [partId : 1 , value = 4567 ])

scala> rdd3.aggregate("")((x,y)=> math.max(x.length,y.length).toString,(x,y)=>x+y)
res13: String = 42

执行过程:
第一个分区:
第一次比较: "" "12" 长度最大值 2 2-->"2"
第二次比较: “2” “23” 长度最大值 2 2-->"2"

第二个分区:
第一次比较: "" "345" 长度最大值 3 3-->"3"
第二次比较: “3” “4567” 长度最大值 4 4-->"4"


2、
rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)

scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res18: String = 11

执行过程:
第一个分区:
第一次比较: "" "12" 长度最小值 0 0-->"0"
第二次比较: “0” “23” 长度最小值 1 1-->"1"

第二个分区:
第一次比较: "" "345" 长度最小值 0 0-->"0"
第二次比较: “0” “4567” 长度最小值 1 1-->"1"

val rdd3 = sc.parallelize(List("12","23","345",""),2)
rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)

scala> val rdd3 = sc.parallelize(List("12","23","345",""),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[10] at parallelize at <console>:27

scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res19: String = 10

scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res20: String = 01

3、aggregateByKey:类似于aggregate,区别:操作的是 key value 的数据类型。

scala> def fun3(index:Int, iter:Iterator[(String,Int)]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }

scala> pairRDD.mapPartitionsWithIndex(fun3).collect
res22: Array[String] = Array(

[partId : 0 , value = (cat,2) ], [partId : 0 , value = (cat,5) ], [partId : 0 , value = (mouse,4) ],


[partId : 1 , value = (cat,12) ], [partId : 1 , value = (dog,12) ], [partId : 1 , value = (mouse,2) ])

(1)将每个动物园(分区)中,动物数最多的动物,进行求和

动物园0
[partId : 0 , value = (cat,2) ], [partId : 0 , value = (cat,5) ], [partId : 0 , value = (mouse,4) ],

动物园1
[partId : 1 , value = (cat,12) ], [partId : 1 , value = (dog,12) ], [partId : 1 , value = (mouse,2) ])

pairRDD.aggregateByKey(0)(math.max(_,_),_+_)

scala> pairRDD.aggregateByKey(0)(math.max(_,_),_+_).collect
res24: Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))

(2)将所有动物求和
pairRDD.aggregateByKey(0)(_+_,_+_).collect

scala> pairRDD.reduceByKey(_+_).collect
res27: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))

aggregateByKey效率更高。

4、 coalesce与repartition
与分区有关
都是对RDD进行重分区。

区别:
coalesce 默认不会进行Shuffle 默认 false 如需修改分区,需置为true
repartition 会进行Shuffle

scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[17] at parallelize at <console>:27

scala> val rdd2 = rdd1.repartition(3)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[21] at repartition at <console>:29

scala> rdd2.partitions.length
res28: Int = 3

scala> val rdd3 = rdd1.coalescse(3,true)
<console>:29: error: value coalescse is not a member of org.apache.spark.rdd.RDD[Int]
val rdd3 = rdd1.coalescse(3,true)
^

scala> val rdd3 = rdd1.coalesce(3,true)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[25] at coalesce at <console>:29

scala> rdd3.partitions.length
res29: Int = 3

scala> val rdd4 = rdd1.coalesce(4)
rdd4: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[26] at coalesce at <console>:29

scala> rdd4.partitions.length
res30: Int = 2

5、其他高级算子
http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html


七、编程案例

(1)分析日志
需求:找到访问量最高的两个网页
(*)第一步:对网页的访问量求和
(*)第二步:排序,降序

(2)创建自定义分区

(3)使用JDBCRDD 操作数据库

(4)操作数据库:把结果存放到数据库中

IDE

-----------------Spark SQL------------------------
类似于Hive

一、Spark SQL 基础

1、什么是Spark SQL
Spark SQL is Apache Spark's module for working with structured data.
Spark SQL 是spark 的一个模块。来处理 结构化 的数据
不能处理非结构化的数据

特点:
1、容易集成
不需要单独安装。

2、统一的数据访问方式
结构化数据的类型:JDBC JSon Hive parquer文件 都可以作为Spark SQL 的数据源
对接多种数据源,且使用方式类似

3、完全兼容hive
把Hive中的数据,读取到Spark SQL中运行。

4、支持标准的数据连接
JDBC

2、为什么学习Spark SQL
执行效率比Hive高

hive 2.x 执行引擎可以使用 Spark

3、核心概念:表(DataFrame DataSet)
mysql中的表:表结构、数据
DataFrame:Schema、RDD(数据)

DataSet 在spark1.6以后,对DataFrame做了一个封装。


4、创建DataFrame
(*)测试数据:员工表、部门表


第一种方式:使用case class

1、定义Schema
样本类来定义Schema。
case class 特点:
可以支持模式匹配,使用case class建立表结构

7521, WARD, SALESMAN,7698, 1981/2/22, 1250, 500, 30

case class Emp(empno:Int,ename:String,job:String,mgr:Int,hiredate:String,sal:Int,comm:Int,deptno:Int)

2、读取文件
val lines = sc.textFile("/usr/local/tmp_files/emp.csv").map(_.split(","))

3、把每行数据,映射到Emp上。
val allEmp = lines.map(x => Emp(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt))

4、生成DataFrame
val df1 = allEmp.toDF

df1.show


第二种方式 使用Spark Session

(1)什么是Spark Session
Spark session available as 'spark'.
2.0以后引入的统一访问方式。可以访问所有的Spark组件。

def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame

(2)使用StructType来创建Schema

val struct =
StructType(
StructField("a", IntegerType, true) ::
StructField("b", LongType, false) ::
StructField("c", BooleanType, false) :: Nil)


case class Emp(empno:Int,
ename:String,
job:String,
mgr:Int,
hiredate:String,
sal:Int,
comm:Int,
deptno:Int)

val myschema = StructType(
List(
StructField("empno",DataTypes.IntegerType),
StructField("ename",DataTypes.StringType),
StructField("job",DataTypes.StringType),
StructField("mgr",DataTypes.IntegerType),
StructField("hiredate",DataTypes.StringType),
StructField("sal",DataTypes.IntegerType),
StructField("comm",DataTypes.IntegerType),
StructField("deptno",DataTypes.IntegerType),
))

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

准备数据 RDD[Row]
val allEmp = lines.map(x => Row(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt))

import org.apache.spark.sql.Row

val df2 = spark.createDataFrame(allEmp,myschema)

第三种方式
直接读取一个带格式的文件。

val df3 = spark.read 读文件,默认是Parquet文件
val df3 = spark.read.json("/usr/local/tmp_files/people.json")

df3.show

val df4 = spark.read.format("json").load("/usr/local/tmp_files/people.json")


5、操作DataFrame

(1)DSL语句
mybatis Hibernate

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

$"sal" 可以看做是一个变量。

查询薪水大于2000的员工
df1.filter($"sal" > 2000).show

求每个部门的员工人数
df1.groupBy($"deptno").count.show

select deptno,count(1) from emp group by deptno

(2)SQL语句

注意:不能直接执行SQL,需要生成一个视图,再执行sql。

scala> df1.create
createGlobalTempView createOrReplaceTempView createTempView

一般用到 createOrReplaceTempView createTempView
视图:类似于表,但不保存数据。

df1.createOrReplaceTempView("emp")

操作:
spark.sql("select * from emp").show

查询薪水大于2000的员工
spark.sql("select * from emp where sal > 2000").show

求每个部门的员工人数
spark.sql("select deptno,count(1) from emp group by deptno").show

(3)多表查询
10,ACCOUNTING,NEW YORK

case class Dept(deptno:Int,dname:String,loc:String)
val lines = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
val allDept = lines.map(x=>Dept(x(0).toInt,x(1),x(2)))

df5.createOrReplaceTempView("dept")

spark.sql("select dname,ename from emp,dept where emp.deptno=dept.deptno").show

6、操作DataSet
跟DataFrame类似,是一套新的接口。高级的Dataframe

举例:
(1)创建DataSet
1、使用序列来创建DataSet。

定义一个case class
case class MyData(a:Int,b:String)

生成序列,并创建DataSet
val ds = Seq(MyData(1,"Tom"),MyData(2,"Mary")).toDS


.toDS 生成DataSet

ds.show


2、使用JSON数据来创建DataSet

定义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

3、使用其他数据

RDD操作和DataFrame操作相结合 ---> DataSet

读取数据,创建DataSet
val linesDS = spark.read.text("/usr/local/tmp_files/test_WordCount.txt").as[String]

对DataSet进行操作:
val words = linesDS.flatMap(_.split(" ")).filter(_.length > 3)

words.show
words.collect

执行一个WordCount程序
val result = linesDS.flatMap(_.split(" ")).map((_,1)).groupByKey( x => x._1).count
result.show

排序:
result.orderBy($"value").show
result.orderBy($"count(1)").show


(2)DataSet操作案例

使用emp.json 生成一个DataFrame
val empDF = spark.read.json("/usr/local/tmp_files/emp.json")

查询工资大于3000的员工
empDF.where($"sal" >= 3000).show

创建case class

case class Emp(empno:BigInt,ename:String,job:String,mgr:String,hiredate:String,sal:BigInt,comm:String,deptno:BigInt)

生成DataSet
val empDS = empDF.as[Emp]

查询工资大于3000的员工
empDS.filter(_.sal > 3000).show

查询10号部门的员工
empDS.filter(_.deptno == 10).show


(3)多表查询

1、创建部门表
val deptRDD = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
case class Dept(deptno:Int,dname:String,loc:String)

val deptDS = deptRDD.map( x=> Dept(x(0).toInt,x(1),x(2))).toDS

2、创建员工表
case class Emp(empno:Int,ename:String,job:String,mgr:Int,hiredate:String,sal:Int,comm:Int,deptno:Int)
val empRDD = sc.textFile("/usr/local/tmp_files/emp.csv").map(_.split(","))

7369,SMITH,CLERK,7902,1980/12/17,800,0,20
val empDS = empRDD.map(x=> Emp(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt)).toDS

3、执行多表查询:等值连接

val result = deptDS.join(empDS,"deptno")

val result1 = deptDS.joinWith(empDS, deptDS("deptno") === empDS("deptno") )

join 和 joinWith 区别:连接后schema不同

4、多表连接后再筛选
deptDS.join(empDS,"deptno").where("deptno == 10").show

7、Spark SQL 中的视图

视图是一个虚表,不存储数据。

两种类型:

1、普通视图(本地视图):只在当前Session中有效。createOrReplaceTempView createTempView

2、全局视图: createGlobalTempView
在不同的Session中都有用 把全局视图创建在命名空间中:global_temp中。类似于一个库。

scala> df1.create
createGlobalTempView createOrReplaceTempView createTempView


举例:
创建一个新session,读取不到emp视图
spark.newSession.sql("select * from emp")

以下两种方式均可读到 全局视图 中的数据。
df1.createGlobalTempView("emp1")
spark.newSession.sql("select * from global_temp.emp1").show

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


二、使用数据源
在Spark SQL中,可以使用各种各样的数据源来操作。 结构化

1、使用load函数、save函数
load函数是加载数据,save是存储数据。

注意:使用load 或 save时,默认是Parquet文件。列式存储文件。

举例:
读取 users.parquet 文件
val userDF = spark.read.load("/usr/local/tmp_files/users.parquet")

userDF.printSchema
userDF.show

val userDF = spark.read.load("/usr/local/tmp_files/emp.json")


保存parquet文件
userDF.select($"name",$"favorite_color").write.save("/usr/local/tmp_files/parquet")

读取刚刚写入的文件:
val userDF1 = spark.read.load("/usr/local/tmp_files/parquet/part-00000-1ab4e661-32c6-441a-b320-79d")---> 不推荐

生产:
val userDF2 = spark.read.load("/usr/local/tmp_files/parquet")


读json文件 必须format
val userDF = spark.read.format("json").load("/usr/local/tmp_files/emp.json")
val userDF3 = spark.read.json("/usr/local/tmp_files/emp.json")

关于save函数:
调用save函数的时候,可以指定存储模式,追加、覆盖等等

userDF2.write.save("/usr/local/tmp_files/parquet")

userDF2.write.save("/usr/local/tmp_files/parquet")
org.apache.spark.sql.AnalysisException: path file:/usr/local/tmp_files/parquet already exists.;

save的时候覆盖
userDF2.write.mode("overwrite").save("/usr/local/tmp_files/parquet")

将结果保存成表
userDF2.select($"name").write.saveAsTable("table1")

scala> userDF.select($"name").write.saveAsTable("table2")

scala> spark.sql("select * from table2").show
+------+
| name|
+------+
|Alyssa|
| Ben|
+------+

2、Parquet文件:列式存储文件,是Spark SQL 默认的数据源

就是一个普通的文件

举例:

1、把其他文件,转换成Parquet文件
调用save函数

把数据读进来,再写出去,就是Parquet文件。

val empDF = spark.read.json("/usr/local/tmp_files/emp.json")
empDF.write.mode("overwrite").save("/usr/local/tmp_files/parquet")
empDF.write.mode("overwrite").parquet("/usr/local/tmp_files/parquet")

val emp1 = spark.read.parquet("/usr/local/tmp_files/parquet")
emp1.createOrReplaceTempView("emp1")
spark.sql("select * from emp1")

2、支持Schema的合并
项目开始 表结构简单 schema简单
项目越来越大 schema越来越复杂

举例:
通过RDD来创建DataFrame
val df1 = sc.makeRDD(1 to 5).map( i => (i,i*2)).toDF("single","double")
"single","double" 是表结构
df1.show

df1.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=1")

val df2 = sc.makeRDD(6 to 10).map( i => (i,i*3)).toDF("single","triple")
df2.show
df2.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=2")

合并两个部分
val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")

val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")


通过RDD来创建DataFrame
val df1 = sc.makeRDD(1 to 5).map( i => (i,i*2)).toDF("single","double")
"single","double" 是表结构
df1.show

df1.write.mode("overwrite").save("/usr/local/tmp_files/test_table/tzkt=1")

val df2 = sc.makeRDD(6 to 10).map( i => (i,i*3)).toDF("single","triple")
df2.show
df2.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=2")

合并两个部分
val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")

val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")

3、json文件

读取Json文件,生成DataFrame
val peopleDF = spark.read.json("/usr/local/tmp_files/people.json")

peopleDF.printSchema

peopleDF.createOrReplaceTempView("peopleView")

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

Spark SQL 支持统一的访问接口。对于不同的数据源,读取进来,生成DataFrame后,操作完全一样。


4、JDBC

使用JDBC操作关系型数据库,加载到Spark中进行分析和处理。

方式一:
val mysqlDF = spark.read.format("jdbc")
.option("url","jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8")
.option("user","root")
.option("password","123456")
.option("driver","com.mysql.jdbc.Driver")
.option("dbtable","emp").load

mysqlDF.show

方式二:
定义一个Properties类

import java.util.Properties
val mysqlProps = new Properties()
mysqlProps.setProperty("user","root")
mysqlProps.setProperty("password","123456")

val mysqlDF1 = spark.read.jdbc("jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8","emp",mysqlProps)

mysqlDF1.show


5、使用Hive

比较常见
(*)spark SQL 完全兼容hive
(*)需要进行配置
拷贝一下文件到spark/conf目录下:
Hive 配置文件: hive-site.xml
Hadoop 配置文件:core-site.xml hdfs-site.xml

配置好后,重启spark

启动Hadoop 与 hive

spark.sql("create table comany.emp_0410(empno Int,ename String,job String,mgr String,hiredate String,sal Int,comm String,deptno Int)row format delimited fields terminated by ','")

三、在IDE中开发Spark SQL

四、性能优化
与RDD类似

1、把内存中缓存表的数据
直接读取内存的值,来提高性能。

RDD中如何缓存:
rdd.cache 或者 rdd.persist

在Spark SQL中,使用SparkSession.sqlContext.cacheTable

spark中所有context对象
1、sparkContext : SparkCore
2、sql Context : SparkSQL
3、Streaming Context :SparkStreaming

统一起来:SparkSession

操作mysql,启动spark shell 时,需要:
./bin/spark-shell --master spark://node3:7077 --jars /usr/local/tmp_files/mysql-connector-java-8.0.11.jar --driver-class-path /usr/local/tmp_files/mysql-connector-java-8.0.11.jar

val mysqlDF = spark.read.format("jdbc").option("driver","com.mysql.jdbc.Driver").option("url","jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8").option("user","root").option("password","123456").option("dbtable","emp").load

mysqlDF.show
mysqlDF.createOrReplaceTempView("emp")

spark.sqlContext.cacheTable("emp") ----> 标识这张表可以被缓存,数据还没有真正被缓存
spark.sql("select * from emp").show ----> 依然读取mysql
spark.sql("select * from emp").show ----> 从缓存中读取数据

spark.sqlContext.clearCache

清空缓存后,执行查询,会触发查询mysql数据库。

2、了解性能优化的相关参数:参考讲义

-----------------Spark Streaming------------------------
流式计算框架,类似于Storm

常用的实时计算引擎(流式计算)
1、Apache Storm:真正的流式计算

2、Spark Streaming :严格上来说,不是真正的流式计算(实时计算)
把连续的流式数据,当成不连续的RDD
本质:是一个离散计算(不连续)

3、Apache Flink:真正的流式计算。与Spark Streaming相反。
把离散的数据,当成流式数据来处理

4、JStorm

一、Spark Streaming基础

1、什么是 Spark Streaming。

Spark Streaming makes it easy to build scalable fault-tolerant streaming applications.
易于构建灵活的、高容错的流式系统。

特点:
1、易用,已经集成到Spark中
2、容错性:底层RDD,RDD本身具有容错机制
3、支持多种语言:Java Scala Python

2、演示官方的Demo
往Spark Streaming中发送字符串,Spark 接收到以后,进行计数
使用消息服务器 netcat Linux自带
yum install nc.x86_64

nc -l 1234

注意:总核心数 大于等于2。一个核心用于接收数据,另一个用于处理数据

在netcat中写入数据 Spark Streaming可以取到

3、开发自己的NetWorkWordCount程序
和Spark Core类似

问题:Hello Hello
Hello World

现在现象:(Hello,2)
(Hello , 1) (World , 1)

能不能累加起来?保存记录下以前的状态?

通过Spark Streaming提供的算子来实现

二、高级特性

1、什么是DStream?离散流
把连续的数据变成不连续的RDD
因为DStream的特性,导致,Spark Streaming不是真正的流式计算

2、重点算子讲解

(1)updateStateByKey
默认情况下,Spark Streaming不记录之前的状态,每次发数据,都会从0开始
现在使用本算子,实现累加操作。

(2)transform

3、窗口操作
窗口:对落在窗口内的数据进行处理,也是一个DStream,RDD

举例:每10秒钟把过去30秒的数据采集过来

注意:先启动nc 再启动程序 local[2]

4、集成Spark SQL : 使用SQL语句来处理流式数据



5、缓存和持久化:和RDD一样

6、支持检查点:和RDD一样

三、数据源
Spark Streaming是一个流式计算引擎,就需要从外部数据源来接收数据

1、基本的数据源
文件流:监控文件系统的变化,如果文件有增加,读取文件中的内容

希望Spark Streaming监控一个文件夹,如果有变化,则把变化采集过来

RDD队列流:可以从队列中获取数据


套接字流:socketTextStream

2、高级数据源
(1)Flume

Spark SQL 对接flume有多种方式:
push方式:flume将数据推送给Spark Streaming

custom sink 模式:比第一种有更好的健壮性和容错性。使用这种方式,flume配置一个sink。

使用官方提供的spark sink组件

需要把 spark-streaming-flume-sink_2.10-2.1.0.jar 拷贝到flume lib下
需要把 spark-streaming-flume-sink_2.10-2.1.0.jar 拷贝到IDE的lib下添加到build path中


(2)Kafka
在讲Kafka时,举例。

四、性能优化的参数

性能优化:
spark submit的时候,程序报OOM错误
程序跑的很慢

方法:调整spark参数
conf.set...

-----------------Spark 调优------------------------

问题:只要会用就可以,为什么还要精通内核源码与调优?

Spark 性能优化概览:

Spark的计算本质是,分布式计算。
所以,Spark程序的性能可能因为集群中的任何因素出现瓶颈:CPU、网络带宽、或者内存。

CPU、网络带宽,是运维来维护的。
聚焦点:内存。

如果内存能够容纳下所有的数据,那就不需要调优了。
如果内存比较紧张,不足以放下所有数据(10亿量级---500G),需要对内存的使用进行性能优化。
比如:使用某些方法减少内存的消耗。

Spark性能优化,主要针对在内存的使用调优。

Spark性能优化的技术:
1、使用高性能序列化类库
2、优化数据结构
3、对于多次使用的RDD进行持久化、checkpoint
4、持久化级别:MEMORY_ONLY ---> MEMORY_ONLY_SER 序列化
5、Java虚拟机垃圾回收调优
6、Shuffle调优,1.x版本中,90%的性能问题,都是由于Shuffle导致的。

其他性能优化:
1、提高并行度
2、广播共享数据
等等。。。

一、诊断Spark内存使用

首先要看到内存使用情况,才能进行针对性的优化。

1、内存花费:

(1)每个Java对象,都有一个对象头,占用16字节,包含一些对象的元信息,比如指向他的类的指针。
如果对象本身很小,比如int,但是他的对象头比对象自己还大。

(2)Java的String对象,会比他内存的原始数据,多出40个字节。
String内部使用的char数组来保存内部的字符串序列,并且还要保存诸如输出长度之类的信息。
char使用的是UTF-16编码,每个字符会占2个字节。比如,包含10个字符的String,2*10+40=60字节

(3)Java中的集合类型,比如HashMap和LinkedList,内部使用链表数据结构。
链表中的每个数据,使用Entry对象包装。
Entry对象,不光有对象头,还有指向下一个Entry的指针,占用8字节。

(4)元素类型为原始数据类型(int),内部通常会使用原始数据类型的包装类型(Integer)来存储元素。

2、如何判断Spark程序消耗内存情况?
预估

(1)设置RDD的并行度。
两种方法创建RDD,parallelize() textFile() 在这两个方法中,传入第二个参数,设置RDD的partition数量。
在SparkConfig中设置一个参数:
spark.default.parallelism
可以统一设置这个application中所有RDD的partition数量

(2)将RDD缓存 cache()

(3)观察日志:driver日志
/usr/local/spark-2.1.0-bin-hadoop2.7/work

19/04/13 22:01:05 INFO MemoryStore: Block rdd_3_1 stored as values in memory (estimated size 26.0 MB, free 339.9 MB)
19/04/13 22:01:06 INFO MemoryStore: Block rdd_3_0 stored as values in memory (estimated size 26.7 MB, free 313.2 MB)

(4)将这个内存信息相加,就是RDD内存占用量。

二、使用高性能序列化类库

1、数据序列化概述

数据序列化,就是将对象或者数据结构,转换成特定的格式,使其可在网络中传输,或存储在内存或文件中。
反序列化,是相反的操作,将对象从序列化数据中还原出来。
序列化后的数据格式,可以是二进制,xml,Json等任何格式。
对象、数据序列化的重点在于数据的交换与传输。

在任何分布式系统中,序列化都是扮演着一个重要的角色。
如果使用的序列化技术,操作很慢,或者序列化后的数据量还是很大,会让分布式系统应用程序性能下降很多。
所以,Spark性能优化的第一步,就是进行序列化的性能优化。

Spark自身默认会在一些地方对数据进行序列化,比如Shuffle。另外,我们使用了外部数据(自定义类型),也要让其课序列化。

Spark本身对序列化的便捷性和性能进行了取舍
默认情况下:Spark倾向于序列化的便捷性,使用了Java自身提供的序列化机制,很方便使用。

但是,Java序列化机制性能不高,序列化速度慢,序列化后数据较大,比较占用内存空间。

2、kryo

Spark支持使用kryo类库来进行序列化。
速度快,占用空间更小,比Java序列化数据占用空间小10倍。

3、如何使用kryo序列化机制
(1)设置Spark conf

bin/spark-submit will also read configuration options from conf/spark-defaults.conf,
in which each line consists of a key and a value separated by whitespace. For example:

spark.master spark://5.6.7.8:7077
spark.executor.memory 4g
spark.eventLog.enabled true
spark.serializer org.apache.spark.serializer.KryoSerializer

(2)使用kryo是,要求需要序列化的类,要提前注册,以获得高性能

conf.registerKryoClasses(Array(classOf[Count],......))

4、kryo类库的优化
(1)优化缓存大小
如果注册的自定义类型,本身特别大(100个字段),会导致要序列化的对象太大。
此时需要对kyro本身进行优化。因为kryo内部的缓存,可能不能存放这么大的class对象。
spark.kryoserializer.buffer.max 设置这个参数,将其调大。

(2)预先注册自定义类型

虽然不注册自定义类型,kryo也可以正常工作,但会保存一份他的全限定类名,耗费内存。
推荐预先注册要序列化的自定义类型。

三、优化数据结构


四、rdd.cache checkpoint

五、持久化级别:MEMORY_ONLY ---> MEMORY_ONLY_SER 序列化

六、Java虚拟机的调优

1、概述

2、Spark GC原理

3、监测垃圾回收

4、优化Executor内存比例

5、Java GC 调优 (-)

七、shuffle原理

1、优化前

2、优化后

八、其他调优
1、提高并行度
2、广播共享数据

---------------Spark MLlib--------------------------