1. Local: 测试环境
2. Standalone: spark自身的一个调度系统
3. Yarn: Spark客户端直接连接yarn, 不需要额外构建spark集群, 有yarn-client和yarn-cluster两种模式. 主要区别在于driver程序的运行节点
4. Mesos: 国内大环境比较少用
Spark任务使用什么进行提交Shell脚本提交
Spark提交作业参数executor-cores 每个executor使用的内核数, 默认为1, 官方建议2-5个
num-executors 启动executor的数量, 默认为2个
executor-memory executor内存的大小, 默认为1G
dirver-cores dirver使用内核书, 默认为1个
driver-memory driver内存大小, 默认512M
任务提交样例:
spark-submit
--master local[5]
--driver-cores 2
--driver-memory 8g
--executor-cores 4
--num-executors 10
--executor-memory 8g
--class PackageName.className xxx.jar
--name "spark job name"
InputPath
OutputPath
Spark的血统概念
RDD在Lineage依赖方面分为两种Narrow Dependencies与Wide Depencies用来解决数据容错时的高效性以及划分任务时起到的重要作用.
作用: 提供了一个抽象模型,将具体的应用逻辑表达为一系列的转换操作,不同的RDD之间的转换操作还可形成依赖关系,进而实现管道化, 避免了中间结果的存储, 大大减低了数据复制, 磁盘和IO的开销
RDD怎么理解RDD是弹性分布式数据结构, 是spark中的最基本数据抽象, 代表一个不可变, 可分区, 可并行计算的集合
Spark的宽窄依赖, stage和task个数stage: 根据RDD之间的依赖关系的不同将Job划分成不同的Stage, 遇到一个宽依赖则划分一个Stage.
Task: stage是一个taskset, 将stage根据分区数划分成一个个task
RDD任务划分Application: 初始化一个SparkContext即生成一个Application
Job: 一个Action算子生成一个Job
stage: 遇到宽依赖就划分一个stage 等于宽依赖的个数+1
Task: stage中最后一个RDD算子的分区个数就是Task个数
列举Spark中的transformation算子- mapmapPartitionsreduceByKeyaggregateByKeycombinByKey
- reducecollectfirsttakeaggregatecountByKeyforeachsaveAsTextFile
- reduceByKeygroupByKey...ByKey
reduceByKey: 按照key进行聚合,在shuffle之前有个combine(预聚合)操作,返回结果RDD
groupByKey: 按照key进行分组,直接进行shuffle
reducebykey比groupbykey建议使用,但是需要注意是否影响业务逻辑
persist的缓存设置级别MEMORY_ONLY 内存
MEMORY_AND_DISK 内存+磁盘
MEMORY_ONLY_SER 内存(序列化)
MEMORY_AND_DIS_SER 内存+磁盘(序列化)
DISK_ONLY 磁盘
Repartition和Coalesce的区别1.两者都是用来改变RDD的partition数量的, repartition底层调用的就是coalesce方法:coalesce(numPartitions,shuffle = true)
2.repartition一定会shuffle,coalesce根据传入的参数来判断是否发生shuffle,一般情况下增大rdd的partition使用repartition,减少partition数量使用coalesce
简述spark的缓存机制都是做RDD持久化的
cache: 内存, 不会截断血缘关系, 使用计算过程中的数据缓存
checkpoint: 磁盘, 截断血缘关系, 在ck之前必须没有任何任务提交才会生效,ck过程会额外提交一次任务. 等Job完成后, 启动专门的Job去完成checkpoint
开发环境: checkpoint+cache
简述spark的共享变量(累加器和广播变量)累加器是spark中提供的一种分布式的变量机制,其原理类似于mapreduce,即分布式的改变,然后聚合这些改变. 累加器的一个常见的用途是在调试作业时对作业执行过程中的事件计数. 而广播变量用来高效分发较大的对象
共享变量出现的原因:
通常在向spark传递函数时,比如使用map()函数或者filter()传条件时,可以使用驱动器程序中定义的变量,但是集群中运行的每个任务都会得到这些变量的一份新的副本,更新这些驱动的值也不会影响驱动器中的对应变量
spark的两个共享变量,累加器和广播变量, 分别为结果聚合与广播两种常见的通信模式突破了这一限制
如果减少Spark运行中的数据库连接数使用foreachPartition代替foreach, 在foreachPartition内获取数据库的连接
如果使用Spakr实现TopN方法1.
1.按照key对数据进行聚合(groupbykey)
2.将value转换为数据,利用scala的sortBy或sortwith进行排序(mapvalues), 数据量太大会OOM
方法2.
1.取出所有的key
2.对key进行迭代,每取出一个key利用spark的排序算子进行排序
方法3.
1.自定义分区器, 按照key进行分区,使用不同的key进到不同的分区
2.对每个分区运用spark的排序算子进行排序
简述sparkstreamint窗口函数的原理窗口函数是在原来定义的sparkstreaming计算批次大小的基础上再次进行封装,每次计算多个批次的数据,同时还需要传递一个滑动步长的参数,用来设置档次计算任务完成之后下一次从什么地方开始
spark开发调优- 避免创建重复的RDD尽可能复用同一个RDD对多次使用的RDD进行持久化避免使用shuffer算子使用map-side预聚合shuffle操作使用高性能算子使用kryo序列化,优化数据结构在段子函数使用外部变量的时候,如果数据量较大,可以使用广播变量, 可以减少网络传输,不使用会占用过的内存从而频繁的GC,极大影响性能
例如:
reduceBykey/aggregatebykey代替groupbykey
mappartitions代替map
foreachpartitions代替foreach
使用fifter后进行coalesce操作
repartitionandsortwithinparititions代替repartition与sort类操作
简述SparkStreaming窗口函数的原理窗口函数就是在原来的定义的SparkStreaming计算批次大小的基础上再次进行封装,每次计算多个批次的数据, 同时还需要传递一个滑动步长的参数,用来设置档次计算任务完成之后下次从什么地方开始计算.
Spark Streaming控制每秒消费数据的速度通过spark.streaming.kafka.maxRatePerPartition参数来设置Spark Streaming从kafka分区每秒拉取的条数
Spark Streaming背压机制把spark.streaming.backpressure.enabled参数设置为true, 开启背压机制后Spark Streaming会根据延迟动态去kafka消费数据,上限由spark.streaming.kafka.maxRatePerPartition参数控制,所有两个参数会一起使用
Spark Streaming 中stage耗时Spark Streaming stage耗时由最慢的task决定, 所以数据倾斜时某个task运行快慢会导致整个spark streaming都运行非常慢
数据倾斜的表现1.Hadoop的数据倾斜表现
有一个或多个reduce卡住, 卡在99.99%, 一直不能结束各种container报错OOM异常的reducer读写的数据量极大, 至少远远的超过其它整行的reducer伴随着数据的倾斜, 会出现任务被kill等各种诡异的表现
2.hive的数据倾斜
一般发生在sql的group by和join on上,并且和数据逻辑绑定比较深
3.spark的数据倾斜
executor lost,OOM,shuffle过程出错Driver OOM单个Executor执行时间特别久,任务卡在某个阶段不能结束正常运行的任务突然失败
