Spark_SQL性能调优
众所周知，正确的参数配置对提升Spark的使用效率具有极大助力，帮助相关数据开发、分析人员更高效地使用Spark进行离线批处理和SQL报表分析等作业。
性能调优选项
| 选型 | 默认值 | 用途 |
|---|---|---|
| spark.sql.codegen | false | 设为 true 时，Spark SQL 会把每条查询词语在运行时编译为 Java 二进制代码。这可以提高大型查询的性能，但在进行小规模查询时会变慢 |
| spark.sql.inMemoryColumnarStorage.compressed | false | 自动对内存中的列式存储进行压缩 |
| spark.sql.inMemoryColumnarStorage.batchSize | 1000 | 列示缓存时的每个批处理的大小。把这个值调大可能会导致内存不够的异常 |
| spark.sql.parquet.compression.codec | snappy | 使用哪种压缩编码器。可选的选项包括 uncompressed/snappy/gzip/lzo |
几种压缩选项的特点
spark.sql.parquet.compressed.codec 默认值为snappy 这个参数代表使用哪种压缩编码器。可选的选项包括uncompressed/snappy/gzip/lzo
uncompressed这个顾名思义就是不用压缩的意思
| 格式 | 可分割 | 平均压缩速度 | 文本文件压缩效率 | Hadoop压缩编解码器 | 纯java实现 | 原生 | 备注 |
|---|---|---|---|---|---|---|---|
| snappy | 否 | 非常快 | 低 | org.apache.hadoop.io.compress.SnappyCodec | 否 | 是 | Snapp有纯java的移植版，但是在Spark/Hadoop中不能用 |
| gzip | 否 | 快 | 高 | org.apache.hadoop.io.compress.GzipCodec | 是 | 是 | |
| lzo | 否 | 非常快 | 中等 | org.apache.hadoop.io.compress.LzoCodec | 是 | 是 | 需要在每个节点上安装LZO |
可选的调优选项
| Property Name | Default | Meaning |
|---|---|---|
| spark.sql.files.maxPartitionBytes | 128 MB | 读取文件时打包到单个分区的最大字节数 |
| spark.sql.files.openCostInBytes | 4 MB | 打开一个文件的估计成本，取决于有多少字节可以被同时扫描，当多个文件放入一个 partition 时使用此方法。最好时过度估计，这样带有小文件的 partition 就会比带有大文件的 partition 快 |
| spark.sql.broadcastTimeout | 300 | 广播连接中等待时的超时时间（以秒为单位） |
| spark.sql.autoBroadcastJoinThreshold | 10 MB | 配置一个表的最大大小，以便在执行连接向所有节点广播，将该值设置为 -1 的话广播将会被禁用 |
| spark.sql.shuffle.partitions | 200 | 配置 partition 的使用数量当 shuffle 数据或聚合数据时 |
代码示例
初始化设置Spark Application配置
构建SparkSession实例对象
import org.apache.commons.lang3.SystemUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SparkSession}
object PerformanceTuneDemo {
def main(args: Array[String]): Unit = {
// 构建SparkSession实例对象，设置相关属性参数值
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.set("spark.sql.session.timeZone", "Asia/Shanghai")
.set("spark.sql.files.maxPartitionBytes", "134217728")
.set("spark.sql.files.openCostInBytes", "134217728")
.set("spark.sql.shuffle.partitions", "3")
.set("spark.sql.autoBroadcastJoinThreshold", "67108864")
// 构建SparkSession实例对象
val spark: SparkSession = SparkSession.builder()
.config(sparkConf)
.getOrCreate()
}
}
到此这篇关于Spark SQL关于性能调优选项详解的文章就介绍到这了,更多相关Spark SQL性能调优内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！
