spark1.x-spark-sql性能优化_cache join

1、cache join和group by 作为变量

2、设置shuffle过程中的并行度

spark.sql.shuffle.partitions SQLContext.setConf()

在hive数据仓库建设过程中 合理设置数据类型 比如能设置为INT的 不要设置为BigInt
减少数据类型导致的内存开销

填写SQL时 尽量给出明确的列名 比如select name from students 不要写select * 的方式

3、并行处理查询结果

Spark SQL查询结果 如果数据量比较大
超过1000条
就不要一次性collect()到Driver在处理
使用foreach()算子 并行处理查询结果

4、缓存表

对于一条SQL语句中可能多次使用到的表 可以对其进行缓存
使用 SQLContext.cacheTable(tableName)
或者DataFrame.cache()即可
Spark SQL会用内存列存储的格式缓存表 仅扫描需要使用的列 并且自动优化压缩 来最小化内存使用和GC开销

SQLContext.uncacheTable(tableName)可以将表从缓存中移除
用SQLContext.setConf()
设置 spark.sqlinMemoryColumnarStorage.batchSize参数 默认10000 可以配置存储的单位

例如有一张hive表叫做activity

1.CACHE TABLE

//缓存全表
sqlContext.sql(“CACHE TABLE activity”)

//缓存过滤结果
sqlContext.sql(“CACHE TABLE activity_cached as select * from activity where …”)
CACHE TABLE是即时生效(eager)的，如果你想等到一个action操作再缓存数据可以使用CACHE LAZY TABLE,这样操作会直到一个action操作才被触发，例如count(*)

sqlContext.sql(“CACHE LAZY TABLE …”)
取消hive表缓存数据

sqlContext.sql(“UNCACHE TABLE activity”)
2.将dataFrame注册成表并缓存

val df = sqlContext.sql(“select * from activity”)
df.registerTempTable(“activity_cached”)
sqlContext.cacheTable(“activity_cached”)

Tip:cacheTable操作是lazy的，需要一个action操作来触发缓存操作。
对应的uncacheTable可以取消缓存

sqlContext.uncacheTable(“activity_cached”)
3.缓存dataFrame

val df = sqlContext.sql(“select * from tableName”)
df.cache()
2.缓存结果

缓存时看到如下提示:

Added rdd_xx_x in memory on …
如果内存不足，则会存入磁盘中,提示如下:

Added rdd_xx_x on disk on …
缓存数据后可以在Storage上看到缓存的数据

cache

3.一些参数

spark.sql.autoBroadcastJoinThreshold
该参数默认为10M,在进行join等聚合操作时，将小于该值的表broadcast到每台worker，消除了大量的shuffle操作。

spark.rdd.compress true
将rdd存入mem或disk前再进行一次压缩，效果显著，我使用cacheTable了一张表，没有开启该参数前总共cache了54G数据,开启这个参数后只34G,可是执行速度并没有收到太大的影响。

spark.sql.shuffle.partitions
这个参数默认为200，是join等聚合操作的并行度，如果有大量的数据进行操作，造成单个任务比较重,运行时间过长的时候，会报如下的错误:

org.apache.spark.shuffle.FetchFailedException: Connection from /192.168.xx.xxx:53450 closed
这个时候需要提高该值。

5、广播join表

spark.sql.autoBroadcastJoinThreshold 默认10485760 10M

参数设置了一个表在join的时候 最大在多大以内 可以被被广播出去

在内存够用的情况下 可以增加其大小

6、钨丝计划

spark.sql.tungsten.enabled 默认是true 自动管理内存

7、其他参数

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.api.java.JavaSQLContext;
import org.apache.spark.sql.api.java.Row;
import org.apache.spark.sql.hive.api.java.JavaHiveContext;


public class PerformanceTuneDemo {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("simpledemo").setMaster("local");
        conf.set("spark.sql.codegen", "false");
        conf.set("spark.sql.inMemoryColumnarStorage.compressed", "false");
        conf.set("spark.sql.inMemoryColumnarStorage.batchSize", "1000");
        conf.set("spark.sql.parquet.compression.codec", "snappy");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaSQLContext sqlCtx = new JavaSQLContext(sc);
        JavaHiveContext hiveCtx = new JavaHiveContext(sc);

        List<Row> result = hiveCtx.sql("SELECT foo,bar,name from pokes2 limit 10").collect();
        for (Row row : result) {
            System.out.println(row.getString(0) + "," + row.getString(1) + "," + row.getString(2));
        }
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

Beeline 命令行设置优化参数
beeline> set spark.sql.codegen=true;
SET spark.sql.codegen=true
spark.sql.codegen=true
Time taken: 1.196 seconds

spark.sql.codegen
Spark SQL在每次执行次，先把SQL查询编译JAVA字节码。针对执行时间长的SQL查询或频繁执行的SQL查询，此配置能加快查询速度，因为它产生特殊的字节码去执行。但是针对很短(1 - 2秒)的临时查询,这可能增加开销,因为它必须先编译每一个查询。

spark.sql.inMemoryColumnarStorage.batchSize：
When caching SchemaRDDs, Spark SQL groups together the records in the RDD in batches of the size given by this option (default: 1000), and compresses each batch. Very small batch sizes lead to low compression, but on the other hand very large sizes can also be problematic, as each batch might be too large to build up in memory.