spark sql优化

1、内存优化

1.1、RDD

RDD默认cache仅使用内存

可以看到使用默认cache时，四个分区只在内存中缓存了3个分区，4.4G的数据

使用kryo序列化+MEMORY_ONLY_SER

可以看到缓存了四个分区的全部数据，且只缓存了1445.8M

所以这两种缓存方式如何选择，官网建议

也就是说集群资源足够使用默认cache，资源紧张使用kryo序列化+MEMORY_ONLY_SER

1.2、DataFrame与DataSet

DataSet不使用Java和Kryo序列化，它使用特殊的编码器序列化

使用默认cache，保存在内存和磁盘

同样多的数据也是全部缓存，只使用了内存612.3M

使用序列化缓存时比使用默认缓存还多缓存了30M，共646.2M

df和ds直接使用默认cache即可

2、小文件过多问题

2.1、RDD中并行度设置

spark.default.parallelism

For distributed shuffle operations like reduceByKey and join, the largest number of partitions in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster manager:
Local mode: number of cores on the local machine
Mesos fine grained mode: 8
Others: total number of cores on all executor nodes or 2, whichever is larger
Default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by user.

2.2、spark sql

Spark SQL can cache tables using an in-memory columnar format by calling sqlContext.cacheTable("tableName") or dataFrame.cache(). Then Spark SQL will scan only required columns and will automatically tune compression to minimize memory usage and GC pressure. You can call sqlContext.uncacheTable("tableName") to remove the table from memory.
建议如果下面无其它任务，缓存可以不释放，有其它任务要释放
算子方式
result.unpersist

spark.sql.shuffle.partitions Configures the number of partitions to use when shuffling data for joins or aggregations.

默认200
如不减少分区，join后hadoop上会有200个小文件

前三个stage为读文件（控制不了），后两个stage为join并行度，为200
1、使用coalesce算子缩小分区，不能大于原有分区数值
2、如果数值小于vcore，有些vcore就不会工作，速度会慢
如压缩成1，并行度就是1只有一个vcore在工作，不会shuffle，如果数据量很大且参数很小，可能会产生oom

可以看到将分区减少到20，hadoop上只有20个文件

3、合理利用cpu资源

未优化时任务可以看到最后200个任务没有平均分到每台机器上，压力全在hadoop103上，如果数据量很少，hadoop103上有些vcore可能没数据在空转，没有合理利用cpu资源

将·spark.sql.shuffle.partitions设置为总vcore的2到3倍可以达到最优效果

不添加缩小分区coalesce可以看到有36个任务
任务分配也很平均，达到优化效果，避免空转情况，合理利用cpu资源，任务时间缩短到2.5分钟

4、广播join

将小表聚合到driver端，分发到每个executor，规避shuffle，避免此stage
只适合小表join大表

正常大表join大表走SortMergeJoin

小于等于10M，自动进行广播join
spark.sql.autoBroadcastJoinThreshold 10485760 (10 MB)

Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1 broadcasting can be disabled. Note that currently statistics are only supported for Hive Metastore tables where the command ANALYZE TABLE <tableName> COMPUTE STATISTICS noscan has been run.

4.1 API

禁用掉广播join ，设置参数为-1