2、尽可能复用同一个RDD
3、对多次使用的RDD进行持久化
//缓存在内存中,因为不进行序列化与反序列化操作,性能最高,要求内存足够大
studentRDD.cache()
studentRDD.persist(StorageLevel.MEMORY_ONLY)
//若内存溢出,可将RDD数据序列化后再保存在内存中
studentRDD.persist(StorageLevel.MEMORY_ONLY_SER)
//将数据序列化后缓存在内存中,内存缓存不下才会写入磁盘
studentRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
4、尽量避免使用shuffle类算子
5、使用map-side预聚合的shuffle操作
当一个小表join大表时,可将小表广播出去,在map端进行关联(小表不超过1G)
6、使用高性能的算子
reduceByKey/aggregateByKey
mapPartitions
foreachPartitions
coalesce
val rePartRDD: RDD[String] = studentsRDD.repartition(10)
println(s"rePartRDD分区数据:${rePartRDD.getNumPartitions}")
val coalesceRDD: RDD[String] = rePartRDD.coalesce(100, shuffle = true)
println(s"coalesceRDD分区数据:${coalesceRDD.getNumPartitions}")
coalesceRDD
.coalesce(1, shuffle = false) //合并小文件
.saveAsTextFile("data/coalesce")
//缓存在内存中,因为不进行序列化与反序列化操作,性能最高,要求内存足够大
studentRDD.cache()
studentRDD.persist(StorageLevel.MEMORY_ONLY)
//若内存溢出,可将RDD数据序列化后再保存在内存中
studentRDD.persist(StorageLevel.MEMORY_ONLY_SER)
//将数据序列化后缓存在内存中,内存缓存不下才会写入磁盘
studentRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)
4、尽量避免使用shuffle类算子
5、使用map-side预聚合的shuffle操作
当一个小表join大表时,可将小表广播出去,在map端进行关联(小表不超过1G)
6、使用高性能的算子
reduceByKey/aggregateByKey
mapPartitions
foreachPartitions
coalesce
val rePartRDD: RDD[String] = studentsRDD.repartition(10)
println(s"rePartRDD分区数据:${rePartRDD.getNumPartitions}")
val coalesceRDD: RDD[String] = rePartRDD.coalesce(100, shuffle = true)
println(s"coalesceRDD分区数据:${coalesceRDD.getNumPartitions}")
coalesceRDD
.coalesce(1, shuffle = false) //合并小文件
.saveAsTextFile("data/coalesce")
当一个小表join大表时,可将小表广播出去,在map端进行关联(小表不超过1G)
6、使用高性能的算子
reduceByKey/aggregateByKey
mapPartitions
foreachPartitions
coalesce
val rePartRDD: RDD[String] = studentsRDD.repartition(10)
println(s"rePartRDD分区数据:${rePartRDD.getNumPartitions}")
val coalesceRDD: RDD[String] = rePartRDD.coalesce(100, shuffle = true)
println(s"coalesceRDD分区数据:${coalesceRDD.getNumPartitions}")
coalesceRDD
.coalesce(1, shuffle = false) //合并小文件
.saveAsTextFile("data/coalesce")
reduceByKey/aggregateByKey
mapPartitions
foreachPartitions
coalesce
7、广播大变量
如果使用的外部变量比较大,建议使用Spark的广播功能,对该变量进行广播。广播 后的变量,会保证每个Executor的内存中,只驻留一份变量副本,而Executor中的 task执行时共享该Executor中的那份变量副本。这样的话,可以大大减少变量副本的数量,从而减少网络传输的性能开销,并减少对Executor内存的占用开销,降低 GC的频率
8、使用Kryo优化序列化性能
使用Kryo序列化器,优化序列化性能
9、优化数据结构
尽量使用字符串替代对象,
使用原始类型(比如 Int、Long)替代字符串,
使用数组替代集合类型,
这样尽可能地减少内存占用 ,从而降低GC频率,提升性能。
10、使用高性能的库fastutil
二、参数优化
spark-submit
--master yarn-cluster
--num-executors = 50 //设定executor的数量
--executor-memory = 4G //设置每个executor的内存大小
--executor-cores = 2 //设置每个executor的核数
--driver-memory = 2G //设置Driver的内存1G-2G
--conf spark.storage.memoryFraction=0.4 //用于缓存的内存占比默认时0.6,如果代码中没有用到缓存 可以将内存分配给shuffle
--conf spark.shuffle.memoryFraction=0.4 //用户shuffle的内存占比默认0.2
--conf spark.locality.wait=10s //再executor中执行前的等待时间 默认3秒
--conf spark.yarn.executor.memoryOverhead=1024 //堆外内存 默认等于堆内存的10%
--conf spark.network.timeout=200s //网络连接的超时时间 默认120s
总的内存=num-executors*executor-memory
总的核数=num-executors*executor-cores
spark on yarn 资源设置标准
yarn的资源一般占总的资源的80%
1、单个任务总的内存和总的核数一般做多在yarn总资源的1/3到1/2之间
2、在上线内再按照需要处理的数据量来合理指定资源 -- 最理想的情况是一个task对应一个core
三、数据倾斜优化
1、使用Hive ETL预处理数据
如果导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀,可以通过Hive来进行数据预处理(即通过Hive ETL预先对 数据按照key进行聚合,或者是预先和其他表进行join)
2、过滤少数导致倾斜的key
将导致数据倾斜的key(对作业的执行和计算结果不是特别重要的话)过滤除去
3、提高shuffle操作的并行度
增加shuffle后的reduce task的数量,减少每一个reduce中分到的数据量,可以缓解数据倾斜
4、双重聚合
第一次是局部聚合,先给每个key 都打上一个随机数,接着对打上随机数后的数据,执行reduceByKey等聚合操作,进行局部聚合;
然后将各个key的前缀给去掉,再次进行全局聚合操作,就可以得到最终结果了
5、将reduce join转为map join
不使用join算子进行连接操作,而使用Broadcast变量与map类算子实现join操作,进而完全规避掉shuffle类的操作,彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过 collect算子拉取到Driver端的内存中来,然后对其创建一个Broadcast变量;接着对另外一个RDD 执行map类算子,在算子函数内,从Broadcast变量中获取较小RDD的全量数据,与当前RDD的每 一条数据按照连接key进行比对,如果连接key相同的话,那么就将两个RDD的数据用你需要的方式连接起来。
6、采样倾斜key并分拆join操作
提高数据本地级别如果使用的外部变量比较大,建议使用Spark的广播功能,对该变量进行广播。广播 后的变量,会保证每个Executor的内存中,只驻留一份变量副本,而Executor中的 task执行时共享该Executor中的那份变量副本。这样的话,可以大大减少变量副本的数量,从而减少网络传输的性能开销,并减少对Executor内存的占用开销,降低 GC的频率
使用Kryo序列化器,优化序列化性能
9、优化数据结构
尽量使用字符串替代对象,
使用原始类型(比如 Int、Long)替代字符串,
使用数组替代集合类型,
这样尽可能地减少内存占用 ,从而降低GC频率,提升性能。
10、使用高性能的库fastutil
二、参数优化
spark-submit
--master yarn-cluster
--num-executors = 50 //设定executor的数量
--executor-memory = 4G //设置每个executor的内存大小
--executor-cores = 2 //设置每个executor的核数
--driver-memory = 2G //设置Driver的内存1G-2G
--conf spark.storage.memoryFraction=0.4 //用于缓存的内存占比默认时0.6,如果代码中没有用到缓存 可以将内存分配给shuffle
--conf spark.shuffle.memoryFraction=0.4 //用户shuffle的内存占比默认0.2
--conf spark.locality.wait=10s //再executor中执行前的等待时间 默认3秒
--conf spark.yarn.executor.memoryOverhead=1024 //堆外内存 默认等于堆内存的10%
--conf spark.network.timeout=200s //网络连接的超时时间 默认120s
总的内存=num-executors*executor-memory
总的核数=num-executors*executor-cores
spark on yarn 资源设置标准
yarn的资源一般占总的资源的80%
1、单个任务总的内存和总的核数一般做多在yarn总资源的1/3到1/2之间
2、在上线内再按照需要处理的数据量来合理指定资源 -- 最理想的情况是一个task对应一个core
三、数据倾斜优化
1、使用Hive ETL预处理数据
如果导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀,可以通过Hive来进行数据预处理(即通过Hive ETL预先对 数据按照key进行聚合,或者是预先和其他表进行join)
2、过滤少数导致倾斜的key
将导致数据倾斜的key(对作业的执行和计算结果不是特别重要的话)过滤除去
3、提高shuffle操作的并行度
增加shuffle后的reduce task的数量,减少每一个reduce中分到的数据量,可以缓解数据倾斜
4、双重聚合
第一次是局部聚合,先给每个key 都打上一个随机数,接着对打上随机数后的数据,执行reduceByKey等聚合操作,进行局部聚合;
然后将各个key的前缀给去掉,再次进行全局聚合操作,就可以得到最终结果了
5、将reduce join转为map join
不使用join算子进行连接操作,而使用Broadcast变量与map类算子实现join操作,进而完全规避掉shuffle类的操作,彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过 collect算子拉取到Driver端的内存中来,然后对其创建一个Broadcast变量;接着对另外一个RDD 执行map类算子,在算子函数内,从Broadcast变量中获取较小RDD的全量数据,与当前RDD的每 一条数据按照连接key进行比对,如果连接key相同的话,那么就将两个RDD的数据用你需要的方式连接起来。
6、采样倾斜key并分拆join操作
尽量使用字符串替代对象,
使用原始类型(比如 Int、Long)替代字符串,
使用数组替代集合类型,
这样尽可能地减少内存占用 ,从而降低GC频率,提升性能。
三、数据倾斜优化 1、使用Hive ETL预处理数据spark-submit
--master yarn-cluster
--num-executors = 50 //设定executor的数量
--executor-memory = 4G //设置每个executor的内存大小
--executor-cores = 2 //设置每个executor的核数
--driver-memory = 2G //设置Driver的内存1G-2G
--conf spark.storage.memoryFraction=0.4 //用于缓存的内存占比默认时0.6,如果代码中没有用到缓存 可以将内存分配给shuffle
--conf spark.shuffle.memoryFraction=0.4 //用户shuffle的内存占比默认0.2
--conf spark.locality.wait=10s //再executor中执行前的等待时间 默认3秒
--conf spark.yarn.executor.memoryOverhead=1024 //堆外内存 默认等于堆内存的10%
--conf spark.network.timeout=200s //网络连接的超时时间 默认120s总的内存=num-executors*executor-memory
总的核数=num-executors*executor-cores
spark on yarn 资源设置标准yarn的资源一般占总的资源的80%
1、单个任务总的内存和总的核数一般做多在yarn总资源的1/3到1/2之间2、在上线内再按照需要处理的数据量来合理指定资源 -- 最理想的情况是一个task对应一个core
2、过滤少数导致倾斜的key如果导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀,可以通过Hive来进行数据预处理(即通过Hive ETL预先对 数据按照key进行聚合,或者是预先和其他表进行join)
3、提高shuffle操作的并行度将导致数据倾斜的key(对作业的执行和计算结果不是特别重要的话)过滤除去
4、双重聚合增加shuffle后的reduce task的数量,减少每一个reduce中分到的数据量,可以缓解数据倾斜
5、将reduce join转为map join第一次是局部聚合,先给每个key 都打上一个随机数,接着对打上随机数后的数据,执行reduceByKey等聚合操作,进行局部聚合;
然后将各个key的前缀给去掉,再次进行全局聚合操作,就可以得到最终结果了
6、采样倾斜key并分拆join操作不使用join算子进行连接操作,而使用Broadcast变量与map类算子实现join操作,进而完全规避掉shuffle类的操作,彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过 collect算子拉取到Driver端的内存中来,然后对其创建一个Broadcast变量;接着对另外一个RDD 执行map类算子,在算子函数内,从Broadcast变量中获取较小RDD的全量数据,与当前RDD的每 一条数据按照连接key进行比对,如果连接key相同的话,那么就将两个RDD的数据用你需要的方式连接起来。



