Spark
2021-01-17 03:33:29 4 举报AI智能生成
登录查看完整内容
spark应用
作者其他创作
大纲/内容
Spark
Spark-Core
Spark
分布式计算框架，可以基于内存处理数据
与MR的区别
1.Spark可以基于内存处理数据，速度快
2.Spark 中有DAG有向无环图执行引擎
4.Spark 是粗立度资源申请，MR是细粒度资源申请
5.SparkShuffle相对于MR的shuffle更灵活
6.
Spark 技术栈
Spark 代码流程
案例：Spark WordCount
java
SparkConf conf = new SparkConf().setMaster("local").setAppName("xx")
JavaSparkContext sc = new JavaSparkContext(conf)
JavaRDD<String> rdd1 = sc.textFile(.)
rdd2.foreach...
sc.stop()
scala
val conf = SparkConf().setMaster("local").setAppName("xxx")
val sc = SparkContext(conf)
result.foreach...
SparkConf
2.设置Spark application 在WEBUI中的名称。
3.设置Spark application运行的资源，例如：内存
SparkContext
SparkContext Spark的上下文，通往Spark集群的唯一通道
代码流程步骤
1.创建SparkConf val conf = new SparkConf().setMaster..setAppName..
2.创建SparkContext val sc = new SparkContext(conf)
3.创建RDD val rdd = sc.textFile(...)
4.对RDD使用Transformation类算子操作
5.对RDD使用Action触发Transformation算子执行
6.sc.stop
Spark运行模式
Local
多用于本地测试，eclipse，IDEA本地开发Spark应用程序使用
Standalone
Spark自带的资源调度框架，支持分布式搭建
Yarn
Hadoop生态圈中的资源调度框架，Spark可以依赖Yarn去调度资源
Messos
资源调度框架
Spark核心RDD
RDD ：弹性分布式数据集
RDD五大特性
1.RDD是由一系列partition组成
2.算子（函数） 是作用在RDD的partition上的
3.RDD之间有依赖关系
5.partition对外提供最佳的计算位置
注意
3.哪里体现了RDD弹性
RDD之间有依赖关系
RDD的分区数可以多可以少
4.哪里体现了RDD的分布式
partition是分布在多个节点上的
RDD是不存数据的，RDD的partition也是不存数据的，partition中存的是计算逻辑
算子
Transformation类算子，转换算子，懒执行，需要Action算子触发
flatMap
map
reduceByKey
sortBy/sortByKey
filter
sample
有无放回随机抽样
可以指定种子参数，每次抽到的数据一样
join
leftOuterJoin
rightOuterJoin
fullOuterJoin
intersection
subtract
union
distinct
map + reduceByKey + map
cogroup
mapPartitions
将RDD中每个分区为单位进行遍历
mapPartitionsWithIndex
repartition
重新分区，可以增大分区也可以减少分区
shuffle类算子，产生shuffle
repartiton(num) = coalesce(num.shuffle = true)
coalesce
重新分区，可以增大分区也可以减少分区，默认是不产生shuffle
如果由少的分区增加到多的分区，还指定不产生shuffle 不起作用
groupByKey
zip
zipWithIndex
foreach
count
会将结果回收到Driver端
first
first = take(1)
会将结果回收到Driver端
take
take(Int)
collect
reduce
foreachPartition
countByKey
countByValue
持久化算子
cache()
默认将数据存储在内存中
cache() 是懒执行的，需要使用action触发执行
cache() = persist() = persist(StoageLevel.MEMORY_ONLY)
persist()
可以手动指定数据的存储级别，persist 默认就是将数据存入内存中（MEMORY_ONLY）
存储级别：
MEMORY_ONLY
MEMORY_ONLY_SER
MEMORY_AND_DISK
MEMORY_AND_DISK_SER
避免使用 DISK_ONLY和“_2”级别
checkpoint()
checkpoint 直接将数据持久化到磁盘，且位置必须指定，当application运行完成之后不会被清空
对RDD使用checkpoint时，可以切断RDD的依赖关系
当Spark application 中 RDD的lineage长，就算复杂时，可以使用checkpint来将数据持久化到磁盘上，这样后面的RDD恢复时就可以基于当前的checkpoint位置恢复
checkpoint 可以管理Spark状态，保存原数据。
checkpoint也是懒执行，需要action触发执行
总结：
cache 和persist注意问题
1.cache 和persist都是懒执行算子，都需要action触发，持久化的最小单位是partition
2.可以对某个RDD使用cache或者persist之后赋值给一个变量，下次使用这个变量就是使用持久化的数据
3.当使用第2中这种方式，cache和persist之后不能紧跟action算子
4.cache和persist的数据位置是由Spark框架管理，当application执行完成之后，这些数据都会被清空
checkpoint执行流程
1.当Spark job执行完成之后，Spark会从后往前回溯，找到checkpointRDD做标记
3.切断之前的RDD依赖关系
优化：对RDD进行checkpoint之前，可以对RDD进行cache，这样回溯完成重新启动job计算checkpointRDD数据时可以从内存中直接获取
Spark集群搭建
Standalone集群搭建
步骤
3.在spark/conf/spark-env.sh 配置
SPARK_MASTER_HOST=node1
SPARK_MASTER_PORT =7077
SPARK_WORKER_CORES=2
SPARK_WORKER_MEMORY=3g
4.将配置好的spark包发送到其他worker节点
5.在Master节点上启动集群，spark/sbin/start-all.sh
6.访问webui :node1:8080
客户端搭建
原封不动将Spark的安装包发送到一台新的节点就可以
Spark 基于Yarn提交任务配置
在客户端节点 spark/conf/spark-env.sh中配置 HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
SparkPI 集群任务提交
./spark-submit --master spark://node1:7077 --class org.apache.spark.examples.SparkPi ../examples/jars/spark-examples_2.11-2.3.1.jar 1000
Spark任务执行
client
命令
./spark-submit --master spark://node1:7077 --class ..jar ....xx
./spark-submit --master spark://node1:7077 --deploy-mode client --class ..jar ....xx
流程
2.客户端向Master申请资源
3.Master会找到满足资源的节点，启动Executor
4.Executor启动之后会反向注册给Driver
注意：
standalone-client模式适用于程序测试，不适用于生产环境个，当在客户端提交多个任务时，会有网卡流量激增问题，在客户端中可以看到task的执行和结果
cluster
./spark-submit --master spark://node1:7077 --deploy-mode cluster --class ..jar ..xxx
1.在客户端提交application，首先客户端向Master申请启动Driver
2.Master收到请求随机在一台Worker节点上启动Driver
3.Driver启动之后，会向Master申请资源，Master收到请求，找到一批满足资源的Worker节点启动Executor
4.Executor启动之后，会反向注册给Driver
5.Driver发送task，监控task，回收结果
standalone-cluster模式适用于生产环境，在客户端提交多个application时，将所有Driver分散到集群中运行的，没有单节点网卡流量激增的问题，在WEBUI中查看task执行和结果
./spark-submit --master yarn --class ..jar .xxx
./spark-submit --master yarn-client --class ...jar xxx
./spark-submit --master yarn --deploy-mode client --class ..jar xxx
2.客户端向RS申请启动ApplicationMaster
4.RS收到请求，返回给AM一批资源NM节点
yarn-client模式适用于程序测试，不适用于生产环境，当在客户端提交多个任务时，会有网卡流量激增问题，在客户端中可以看到task的执行和结果
./spark-submit --master yarn-cluster --class ...jar ..xxx
./spark-submit --master yarn --deploy-mode cluster --class ...jar ..xx
3.AM启动之后，向RS申请资源用于启动Executor
yarn-cluster模式适用于生产环境，在客户端提交多个application时，将所有Driver分散到集群中运行的，没有单节点网卡流量激增的问题，在WEBUI中查看task执行和结果
Spark 术语
任务层面：Application -> job -> stage -> tasks
资源层面：Master -> Worker -> Executor -> ThreadPool
Spark RDD宽窄依赖
RDD窄依赖
父RDD与子RDD的partition之间的关系是一对一
父RDD与子RDD的partition之间的关系是多对一
RDD宽依赖（shuffle）
父RDD与子RDD的partition之间的关系是一对多
Stage
Stage是由一组并行的task组成
Stage是按照job中RDD的宽窄依赖关系划分，有宽依赖就划分stage
Stage的计算模式 pipeline管道计算模式
stage中的数据什么时候落地
1.shuffle write
2.对RDD进行持久化
Stage的并行度是由Stage中finalRDD的partition个数决定的
提高stage的并行度？
增大RDD的分区数
例如
groupByKey(num)
repartitioin/coalesce
Spark的资源调度和任务调度
资源调度
1.集群启动，Worker向Master汇报资源，Master掌握了集群资源
3.TaskScheduler 向Master申请资源
4.Master找到满足资源的节点，启动Executor
任务调度
6.当RDD有action算子触发生成job时，job中RDD依赖关系形成DAG有向无环图，提交给DAGScheduler来划分
9.Driver监控task执行，回收结果
问题：
1.TaskScheduler会重试失败的task，重试3次，3次之后由DAGScheduler来重试stage，重试4次之后如果task依然失败，当前stage所在的job就失败了，如果application中有一个job失败，那么application就失败了
Spark粗粒度资源申请和细粒度资源申请
粗粒度资源申请--Spark
Application执行之前，先将所有资源申请完毕，task执行和job执行就不需要自己重新申请资源，task和job执行快，当最后一个task执行完成之后才会释放这批资源
优点：application执行快
缺点：集群资源不能充分利用
细粒度资源申请 -- MR
Application执行之前，每个job都会自己重新申请资源，job执行完之后就会释放这批资源
优点：集群资源可以充分利用
缺点：application 执行相对慢
PV&UV
PV:page view
UV:unquie Vistor
Spark提交任务参数
--conf
--master
--name
--jars
--driver-class-path
--files
--driver-cores
--driver-memory
--executor-cores
--executor-memory
--total-executor-cores
--executor-num
--supvise 当Driver失败，会自动重启Driver
Spark源码
Spark 资源调度源码
Spark资源调度结论
1.Executor在集群中分散启动，利于数据处理的本地化
4.可以通过指定 --total-executor-core 来指定当前应用程序使用多少core
5.启动Executor不仅和core有关还和内存有关。
Spark资源调度结论验证
Spark 任务调度源码
从Action算子开始
二次排序
Spark中大于两列的排序都叫二次排序
封装对象，对象中实现Compare 接口，实现对象的排序
分组取topN
原生的集合排序
定长数组
广播变量
当Executor端使用到Driver端的变量时，如果使用广播变量，在每个Executor端都有一份Driver端的变量副本
当Executor端使用到Driver端的变量时，如果不使用广播变量，在每个Executor端有多少task就有多少变量副本
1.RDD不能广播出去，可以将RDD的结果广播出去
2.广播变量在Driver端定义，在Executor端不能改变广播变量的值
累加器
相当于集群中的统筹的变量
LongAccumulator
DoubleAccumulator
自定义累加器（项目）
Spark WEBUI
如何查看task数据倾斜
job -> stage -> tasks
端口
8080
8081
4040
18080
50070
8020
9000
9083
2181
9092
Spark 历史日志服务器
配置步骤：
1.在客户端中../conf/spark-defaults.conf文件中配置
spark.eventLog.dir hdfs://node1:9000/spark/test
spark.eventLog.enabled true
spark.history.fs.logDirectory hdfs://node1:9000/spark/test
spark.eventLog.compress true
3.启动历史日志服务器：../sbin/start-history-server.sh
4.查看webui :node4:18080
Spark Master HA
Master HA是针对Standalone集群，有两种方式
1.基于文件系统
缺点：手动恢复HA
2.基于zookeeper
存储
自动选举
使用zookeeper配置Master HA
1.在Master节点../conf/spark-env.sh 中配置
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.dir=/sparkmaster0821"
2.将配置好的文件发送到其他节点 scp ./spark-evn.sh node2:`pwd`
3.找一台StandBy-Master 节点 ，配置Master ： conf/spark-evn.sh 配置 SPARK_MASTER_HOST=node2
4.在Alive-Master节点启动集群： sbin/start-all.sh
5.在StandBy-Master节点上启动Master : start-master.sh
6.验证
集群已经提交的任务和将要提交的任务都不会受影响
Spark Shuffle
HashShuffleManager
普通机制
产生磁盘小文件的个数：M*R
过程
1.map处理完数据之后写往一个buffer缓冲区（默认与reduce task个数一致）
2.buffer默认32k，满了之后会溢写磁盘，每个buffer对应一个磁盘小文件
3.reduce 拉取数据
问题
产生磁盘小文件多
shuffle write对象多
shuffle read 对象多
shuffle 磁盘文件多，导致节点之间连接多，连接多受到网络不稳定导致的task失败多
优化机制
产生磁盘小文件个数：C*R
1.一个core中map task处理完的数据写往一个buffer缓冲区
相对于普通机制产生的磁盘小文件个数少，如果reduce task 个数多，core多，产生的磁盘小文件也比较多
SortShuffleManager
产生磁盘小文件的个数：2*M
1.map task处理完数据之后，写往一个5M的内存数据结构
2.SortShuffle中会有估算内存机制，如果估算内存不够 ，会申请内存： 申请 = 估算*2 - 当前
3.申请不到内存，会溢写磁盘，溢写磁盘过程中有排序，最终形成2个磁盘文件，一个索引文件，一个数据文件
4.reduce 拉取数据
bypass机制
产生的磁盘小文件个数：2*M
相对于普通机制，只是在溢写磁盘的时候没有排序
触发bypass机制的条件
1.map 端不能有预聚合
2.reduce task 的个数必须小于spark.shuffle.sort.bypassMergeThreshold=200 参数
Shuffle文件寻址
MapOutputTracker
MapOutputTrackerMaster(Driver)
MapOutputTrackerWorker(Executor)
BlockManager
BlockManagerMaster(Driver)
DiskStore
MemoryStroe
BlockTransferService
BlockManagerSlave(Executor)
获取磁盘文件过程‘
2.reduce task处理数据之前向Driver获取磁盘数据位置
3.reduce 端通过BlockManager连接数据所在节点，通过BlockTransferService拉取数据
4.BlockTransferService 一次启动5个task拉去数据，一次拉取48M，拉去数据存在Executor的shuffle聚合内存中
reduce oom?
增大Executor内存
增大shuffle内存比例
减少拉取的数据量
Spark内存管理
Spark1.6之前使用的是静态内存管理，Spark1.6之后使用的是统一内存管理，可以通过参数spark.memory.useLegacyMode 切换
静态内存
0.2
0.6
统一内存
300M
(总-300M) *0.25
(总-300M)*0.75
Shuffle调优
spark.shuffle.file.buffer
spark.reducer.maxSizeInFlight
spark.shuffle.io.maxRetries
spark.shuffle.io.retryWait
spark.shuffle.memoryFraction |spark.memory.fraction
spark.shuffle.sort.bypassMergeThreshold-
spark.shuffle.consolidateFiles
spark.shuffle.manager
SparkSQL
Spark SQL支持Spark中可以对分布式数据使用sql查询
Hive -> Shark -> SparkSQL
Shark & SparkSQL
3.还可以将DataFrame转换成RDD使用
4.SparkSQL相对于Shark来说，完全脱离了Hive的限制
Spark on Hive-SparkSQL
Hive 存储
Spark：解析优化，执行引擎
Hive on Spark-Shark
Hive :解析优化 ，存储
Spark：执行引擎
DataFrame
DataFrame = Dataset<Row>
DataFrame 像一张二维表格，有数据还有列的schema信息
Dataset和RDD相比，Dataset的效率高
谓词下推
创建DataFrame
1.读取Json 格式的文件
sparkSession.read.json...
sparkSession.read.format("json").load...
列会按照Ascii 码排序
dataFrame.show(num)
dataFrame.printSchema
创建临时表的方式
dataFrame.createOrReplaceTempView ...
dataFrame.createOrReplaceGlobalTempView ...(可以跨session)
将DataFrame转换成RDD: val rdd : RDD[Row] = dataFrame.rdd
row.getAs[String](index)
row.get(“name”)
2.读取Json格式的RDD或者DataSet
导入隐式转换，才能使用 dataSet.toDF
import sparkSession.implicits._
3.读取普通的RDD加载DataFrame
通过反射的方式
1.首先需要将数据文件加载成RDD|Dataset
2.将RDD或者Dataset 装换成 对象类型的RDD|DataSet
3.RDD.toDF() | Dataset.toDF()
注意：自动将对象中的属性当做schema中的列，对象中属性的类型当做schema中列的类型
动态创建Schema
1.首先需要将数据文件加载成val rdd = RDD<Row>
4.读取Parquet文件 加载DataFrame
读取parquet文件 方式
sparkSession.read.parquet(...)
sparkSession.read.format("parquet").load(...)
可以将数据保存到parquet格式的文件中
5.读取MySQL 中的数据加载DataFrame
加载数据三种方式
2.spark.read.format("jdbc").options(map).load
3.spark.read.format("jdbc").option().option..load()
将DataFrame结果保存到MySql中
参数：spark.sql.shuffle.partitions 指定sql执行时，解析成Sparkjob 的分区数，默认200
6.读取Hive中的数据加载DataFrame
加载Hive中的数据
sparkSession.enableHiveSupport()
sparkSession.sql("Hive 语句")
sparkSession.table("tableName")
将dataFrame存储在Hive中
dataframe.write.mode(SaveMode.Overwrite).saveAsTable("tabelName")
Spark on Hive配置
在Spark 客户端 ../conf中配置hive-site.xml
<configuration>
<property>
<name>hive.metastore.uris</name>
<value>thrift://node1:9083</value>
</property>
</configuration>
测试：使用MR和SparkSQL 测试对同一批数据查询速度
UDF
User defined Function -- 用户自定义函数
使用UDF是一对一的关系，读取一条数据处理得到一条数据
注册UDF:
使用
UDAF
User defined aggregate Function -- 用户自定义聚合函数
开窗函数
row_number() over(partition by xxx order by xxx)
相当于Spark core中分组取topn
在Hive中使用开窗函数，在Mysql8之后也是支持的
SparkStreaming
SparkStreaming 流式处理模块，7*24小时不间断运行
SparkStreaming & Storm
1.SparkStreaming微批处理，Storm纯实时处理数据，SparkStreaming的吞吐量大
2.SparkStreaming擅长处理复杂的业务，可以在Streaming中使用core ，sql 处理数据，Strom擅长处理简单的汇总型业务
3.Strom的事务相对完善，SparkStreaming目前也是比较完善
4.Storm支持动态资源调度，Spark1.2之后也是支持动态资源调度，不建议开启
SparkStreaming读取Socket数据
nc - lk 9999
本地读取Scokect数据，底层采用了Receiver接受器模式，需要一个task用来接收 数据， local[2]
batchInterval 相当于我们可以接受数据的延迟度。可以通过webui多次调节得到
SparkStreaming.start()之后不能添加新的业务逻辑
SparkStreaming.stop()之后不能重新调用start() 方法重新启动
DStream有自己的Transformation类算子，懒执行，需要DStream的outputOperator类算子触发执行
SparkStreaming 监控目录中的数据
不需要local[2]
目录中的数据必须是原子性产生，已将存在的文集不能被监控到
SparkStreaming 算子
Transformation
filter..
map..
flatMap...
transform
可以获取DStream中的RDD，最终需要返回一个RDD，返回的RDD被封装到一个DStream中
对获取的RDD可以使用RDD的transformation类算子，但是一定要返回RDD
transform算子内，获取的RDD的算子外的代码是在Driver端执行，可以通过这个特点做到动态改变广播变量的值
updateStateByKey
自从SparkStreaming启动以来，根据key更新状态
多久将内存中的状态保存到checkpoint中？
batchInterval >10 batchInterval 保存一次
batchInterval <10 10s保存一次
reduceByKeyAndWindow
窗口长度:wl，窗口长度必须是batchInterval的整数倍
滑动间隔：si，滑动间隔必须是batchInterval的整数倍
每次计算，将窗口内的批次重新计算
需要设置checkpoint保存上一次所有key的状态
window
窗口函数
val ds:DStream[xx] = window(窗口长度，滑动间隔)
outputOperator
foreachRDD
可以获取DStream中的RDD，一定要对获取的RDD使用Action算子触发
foreachRDD算子内，获取的RDD的算子外的代码是在Driver端执行，可以通过这个特点做到动态改变广播变量的值
print
saveAsTextFile
Driver HA
当SparkStreaming应用程度停止之后，可以从checkpoint中恢复数据处理的位置（offset）
checkpoint中存储的数据
conf配置信息
DStream处理逻辑
消费数据的位置(offset)
问题：当代码逻辑变化时，既想从checkpoint中恢复offset信息，又想执行新的逻辑，是不能实现的。因为当恢复offset时，将旧的代码逻辑也恢复过来了
Kafka
术语
kafka是分布式的消息系统，将数据直接存储磁盘，默认保存7天
producer
消息的生产者，自己决定将数据存在哪个partition中，两种机制1.轮询2key的hash
broker
组成kafka集群的节点，之间没有主从关系，依赖zookeeper来协调，broker负责消息的读写，和存储
topic
一类消息，由 partition组成，可以创建topic时指定。topic中的partition目的是为了并行消费topic数据
partition
组成topic的单元，每个partition对应磁盘目录
每个partition有副本，可以在创建topic时指定
每个partition都由一个broker管理，这个broker叫做当前parititon的leader
每个partition只能被同一个消费者组内的一个消费者同时连接消费
consumer
每个消费者都有自己的消费者组
不同的组之间消费相同的topic互不影响
同一个组内的不同消费者消费同一个topic时，这个topic中相同的数据只能被消费一次
kafka 0.8.2 之前 consumer自己将消费者offset维护在zookeeper中
kafka0.9 + consumer 的offset是由kafka维护
zookeeper
kafka 0.8.2 之前存储消费者offset
创建topic
查看topic
console当做生产者
./kafka-console-producer.sh --topic topic2017
console当做消费者
查看某个topic的详细描述
删除topic
在kafka broker节点 ../config/server.properties中配置 topic.delete.enable = true
kafka Leader均衡机制
SparkStreaming + kafka 整合
Spark1.6 + kafka 0.8.2
Spark2.3 + kafka0.11
0 条评论
回复 删除
下一页
