SparkSQL調優

1、執行計劃（過往記憶https://www.iteblog.com/archives/2562.html）

df.explain(true)//顯示邏輯計劃和物理計劃，不加true只顯示物理計劃

2、邏輯計劃優化方法：

謂詞下推，列裁剪，常量替換，常量累加

?

3、優化方法

數據源方面：

1、hive 使用parquet格式，不要用textfile。列式存儲便于查詢引擎做謂詞下推、更優的壓縮算法（不同列可以采取不同的壓縮算法）減少IO，塊遍歷等優化方法。

2、Kafka根據key的hash值分區，OGG到Kafka 表名作為key，因此不同大小的表可以更改表名，均衡分到不同partition。

sparkSQL程序方面(spark優化)：

1、多次用到的表，做cache。默認進行壓縮。

spark.sql.inMemoryColumnarStorage.compressed //默認為true，為每個列選擇壓縮方式
spark.sql.inMemoryColumnarStorage.batchSize //默認為10000 byte 控制列緩存的批量大小。批次大有助于改善內存使用和壓縮，但是緩存數據會有OOM的風險

2、小于10M的表會自動broadcast，走broadcast join，調高廣播表的大小，使其走broadcast join ，但是太大有可能driver端OOM，-1為禁止自動廣播。

當使用的外部變量較大時，也可把外部變量作為廣播變量進行廣播。

spark.sql.autoBroadcastJoinThreshold //默認10485760 (10 MB)

val listBrodcast = spark.sparkContext.broadcast(list)

3、sparkSQL shuffle read partition默認為200，提高可解決部分數據傾斜問題。

spark.sql.shuffle.partitions //默認200

4、讀不可分割的文件，一個文件一個partition，若小文件過多，影響效率，設置多個文件寫入一個分區

spark.sql.files.openCostInBytes //默認4194304 (4 MB)，打開一個文件的時間可讀取4MB數據，因此小于4M的文件讀入一個分區（待驗證）

spark.sql.files.maxPartitionBytes //默認134217728 (128 MB)，文件傳入一個分區里的最大字節數

?5、使用高效的算子

1、reduceByKey/aggregateByKey代替groupByKey//前者partition內部會進行預聚合，后者不進行預聚合直接全局shuffle
2、mapPartitions代替map，foreachpartitions 代替foreach//前者會一次性讀取整個partition的數據進行處理，比如建立數據庫連接在foreachpartitions中，不要在foreach
3、filter之后coallease

6、修改序列化器為kryo，并注冊序列化類

conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.registerKryoClasses(Array(classOf[MyClass1],?classOf[MyClass2])

?7、join方式(https://www.cnblogs.com/suanec/p/7560399.html)