spark SQL 任务参数调优1

1.背景

要了解spark参数调优,首先需要清楚一部分背景资料Spark SQL的执行原理,方便理解各种参数对任务的具体影响。



一条SQL语句生成执行引擎可识别的程序,解析(Parser)、优化(Optimizer)、执行(Execution) 三大过程。其中Spark SQL 解析和优化如下图

  1. Parser模块:未解析的逻辑计划,将SparkSql字符串解析为一个抽象语法树/AST。语法检查,不涉及表名字段。

  2. Analyzer模块:解析后的逻辑计划,该模块会遍历整个AST,并对AST上的每个节点进行数据类型的绑定以及函数绑定,然后根据元数据信息Catalog对数据表中的字段和基本函数进行解析。

  3. Optimizer模块:该模块是Catalyst的核心,主要分为RBO和CBO两种优化策略,其中RBO是基于规则优化(谓词下推(Predicate Pushdown) 、常量累加(Constant Folding) 、列值裁剪(Column Pruning)),CBO是基于代价优化。

  4. SparkPlanner模块:优化后的逻辑执行计划OptimizedLogicalPlan依然是逻辑的,并不能被Spark系统理解,此时需要将OptimizedLogicalPlan转换成physical plan(物理计划),如join算子BroadcastHashJoin、ShuffleHashJoin以及SortMergejoin 。

  5. CostModel模块:主要根据过去的性能统计数据,选择最佳的物理执行计划。这个过程的优化就是CBO(基于代价优化)。

在实际Spark执行完成一个数据生产任务(执行一条SQL)的基本过程:

(1)对SQL进行语法分析,生成逻辑执行计划
(2)从Hive metastore server获取表信息,结合逻辑执行计划生成并优化物理执行计划
(3)根据物理执行计划向Yarn申请资源(executor),调度task到executor执行。
(4)从HDFS读取数据,任务执行,任务执行结束后将数据写回HDFS。

上述运行过程
过程 (2)主要是driver的处理能力
过程 (3)主要是executor 、driver的处理能力、作业运行行为

本文从作业的运行过程(2)(3)各选择一个参数介绍从而了解运行过程。

目前的spark参数以及相关生态的参数列表几百个:
Hadoop参数:https://hadoop.apache.org/docs/current/hadoop-mapreduce-client/hadoop-mapreduce-client-core/mapred-default.xml
hive参数:Configuration Properties - Apache Hive - Apache Software Foundation
spark参数:spark 配置参数 Configuration - Spark 3.5.0 Documentation
                     spark 优化参数 Performance Tuning - Spark 3.5.0 Documentation
                     spark 执行参数 Spark SQL and DataFrames - Spark 2.0.0 Documentation

                     各个公司自定义参数:set spark.sql.insertRebalancePartitionsBeforeWrite.enabled = true

其他网上参考的参数:Hive常用参数总结-CSDN博客

参数列表

参数类型

参数

设置值

描述

资源利用

spark.driver.memory
spark.driver.cores
spark.driver.memoryOverhead
spark.executor.memory

5g

--driver-memory 5G

每个exector的内存大小,后缀"k", "m", "g" or "t"

input split

spark.hadoop.hive.exec.orc.split.strategy

spark.hadoop.mapreduce.input.fileinputformat.split.maxsize;

spark.hadoop.mapreduce.input.fileinputformat.split.minsize;
spark.sql.files.maxPartitionBytes

BI 、ETL 、HYBRID

shuffle

spark.sql.shuffle.partitions

200

spark.default.parallelism

80, 100, 200, 300

join

1.spark.hadoop.hive.exec.orc.split.strategy 参数


            1. 参数作用:参数控制在读取ORC表时生成split的策略,影响任务执行时driver压力和mapper 数量。
            2. 参数介绍 : 参数来源于hive  :hive.exec.orc.split.strategy官方定义如下图,当任务执行开始时,ORC有三种分割文件的策略 BI 、ETL 、HYBRID(默认)
HYBRID模式:文件数过多和文件小的场景下,当文件数大于mapper count (总文件大小/hadoop默认分割大小128M) 且文件大小小于HDFS默认(128M)的大小。
ETL:生成分割文件之前首先读取ORC文件的footer(存储文件信息的文件),
BI: 直接分割文件,没有访问HDFS上的数据。

ORC文件的footer是什么?
  ORC 文件原理:全称 Optimized Row Columnar 1.ORC是一个文件格式比较高效的读取、写入、处理hive数据。(我之前理解是一个高效压缩文件)。2.序列化和压缩: intger和String 序列化。按照文件块增量的压缩。
文件结构:三级结构:stripes 存在具体的数据行组(索引、数据行、stripe footer 的信息),file footer 文件的辅助信息(stripe的列表、每个stripe行数、列的数据类型、列上聚合信息 最大值最小值),psotscipt 文件的压缩参数和压缩后的大小。

       3.使用方法和场景: 因此ETL模式下读取的file footer是每个orc文件块的辅助信息。对于一些较大的ORC表,footer可能非常大,ETL模式下读取大量hdfs的数据信息切分文件,导致driver的开销压力过大,这种情况适用BI模式比较合适。
    一些配合使用参数 如:spark.hadoop.mapreduce.input.fileinputformat.split.maxsize; spark.hadoop.mapreduce.input.fileinputformat.split.minsize; map输入最小最大分割块,maxsize 和minsize在输入端控制ORC文件的分割合并。当spark 从hive表中读取数据是会创建一个HadoopRDD的实例,HadoopRDD根据computeSplitSize方法分割文件(org.apache.hadoop.mapreduce.lib.input.FileInputFormat ) Math.max(minSize, Math.min(maxSize, blockSize) 源代码Source code,因此文件表的小文件过多3M大小,根据公式一个小文件就是一个split分割生成大量的patitions,导致tasks数量就巨大,整个任务性能瓶颈可能在读取资源数据缓解。

文件分割源码

   spark.sql.files.maxPartitionBytes  单partition的最大字节数, 为了防止把已经设置好的分割块再次合并,可以将 set spark.hadoopRDD.targetBytesInPartition=-1。

2.spark.sql.shuffle.partitions


    参数作用: 在任务有shuffle时候(join或者聚合场景下)控制partitions的数量。
    参数介绍:

Property Name

Default

meaning

链接

翻译

不同点

共同点

spark.sql.shuffle.partitions

200

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

Spark SQL and DataFrames - Spark 2.0.0 Documentation

Spark SQL中shuffle过程中Partition的数量

仅适用于DataFrame ,group By, join 触发数据shuffle,因此这些数据转换后的结果会导致分区大小需要通过Spark.sql.shuffle.partitions 中设置的值。
如果任务没有join 或者聚合操作,参数设置不会生效。

配置shuffle partitions 的数量

spark.default.parallelism

For distributed shuffle operations like reduceByKeyand join, the largest number of partitions in a parent RDD. For operations like parallelizewith 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.

Configuration - Spark 3.5.0 Documentation

1.reduceByKey
指定分区数  val rdd2 = rdd1.reduceByKey( _ + _, 10)
不指定分区数val rdd1 = rdd2.reduceByKey(_ + _ )
2.join 
val rdd3 = rdd1.join(rdd2),rdd3里Partition的数量由父rdd中最多的Partition数量决定,因此使用join算子时,应增加父rdd中的Partition数量。

1.若当前RDD执行shuffle操算子如reducebykey 和join ,则为在父RDD中最大的partition数。
2.若当前RDD没有上一个RDD则集群管理器分配
  2.1 本地模式:机器核数
  2.2 Mesos上 8
  2.3 所有executor的核数或者是2的最大值

spark.default.parallelism 是随 RDD 引入的,当用户未设置时候,返回reduceByKey(), groupByKey(), join() 转换的默认分区数,仅适用于RDD。

参数用法:在提交作业的通过 --conf 来修改这两个设置的值,方法如下:或者
         spark-submit --conf spark.sql.shuffle.partitions=300 --conf spark.default.parallelism=300
                     sqlContext.setConf("spark.sql.shuffle.partitions", "300")
                     sqlContext.setConf("spark.default.parallelism", "300”)

参数介绍2.0:chatGPT3.5 的答案

     理解spark的并行度:

  1.  资源的并行  exector数和cpu core数

  2.  数据的并行  spark作业在各个stage的task 的数量是并行执行,task数量设置成Spark Application总CPU core数量的2~3倍,同时尽量提升Spark运行效率和速度;

    
     扩展: flink 的并行度

参考文档:
1.Spark SQL底层执行流程详解(好文收藏)-腾讯云开发者社区-腾讯云  spark 执行原理
2.ORC 参数:Configuration Properties - Apache Hive - Apache Software Foundation
3.ORC文件定义: LanguageManual ORC - Apache Hive - Apache Software Foundation
4.oRC解读: 深入理解ORC文件结构-CSDN博客
5.hadoop input: How does Spark SQL decide the number of partitions it will use when loading data from a Hive table? - Stack Overflow
6.文件分割:从源码看Spark读取Hive表数据小文件和分块的问题 - 掘金, How does Spark SQL decide the number of partitions it will use when loading data from a Hive table? - Stack Overflow
7.spark手册:How to Set Apache Spark Executor Memory - Spark By {Examples}
8.并行: performance - What is the difference between spark.sql.shuffle.partitions and spark.default.parallelism? - Stack Overflow
9.flink的并行 : 并行执行 | Apache Flink
10.reducebykey :scala - reduceByKey: How does it work internally? - Stack Overflow
11.key values : 4. Working with Key/Value Pairs - Learning Spark [Book]
12.spark并行:    Spark调优之 -- Spark的并行度深入理解(别再让资源浪费了)_spark并行度-CSDN博客
13.场景:  spark SQL 任务参数调优1

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/123805.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

106.从中序与后序遍历序列构造二叉树

力扣题目链接(opens new window) 根据一棵树的中序遍历与后序遍历构造二叉树。 注意: 你可以假设树中没有重复的元素。 例如,给出 中序遍历 inorder [9,3,15,20,7]后序遍历 postorder [9,15,7,20,3] 返回如下的二叉树: class Solution { public:Tr…

大语言模型之十四-PEFT的LoRA

在《大语言模型之七- Llama-2单GPU微调SFT》和《大语言模型之十三 LLama2中文推理》中我们都提到了LoRA(低秩分解)方法,之所以用低秩分解进行参数的优化的原因是为了减少计算资源。 我们以《大语言模型之四-LlaMA-2从模型到应用》一文中的图…

AJAX--Express速成

一、基本概念 1、AJAX(Asynchronous JavaScript And XML),即为异步的JavaScript 和 XML。 2、异步的JavaScript 它可以异步地向服务器发送请求,在等待响应的过程中,不会阻塞当前页面。浏览器可以做自己的事情。直到成功获取响应后&#xf…

【Spring Cloud】深入探索统一网关 Gateway 的搭建,断言工厂,过滤器工厂,全局过滤器以及跨域问题

文章目录 前言为什么需要网关以及网关的作用网关的技术实现 一、Gateway 网关的搭建1.1 创建 Gateway 模块1.2 引入依赖1.3 配置网关1.4 验证网关是否搭建成功1.5 微服务结构分析 二、Gateway 断言工厂2.1 Spring 提供的断言工厂2.2 示例:设置断言工厂 三、Gateway …

RabbitMQ安装与简单使用

安装 下载资源 可以访问官网查看下载信息rabbitmq官网 选择合适的版本,注意:rabbitmq需要下载一个Erlang才能使用 我自己是在一下两个连接中下载的 rabbitmq 3.8.8 erlang 21.3.8.15 需要下载其他版本的同学注意erlang版本是否匹配,可以访…

Java Stream的基本使用

Stream特点 Stream的一系列操作组成了Stream的流水线, Stream流水线包含: 数据源: 这里的数据源可能是集合/数组, 可能是生成器, 甚至可能是IO通道(Files.lines)零个或多个中间操作: 中间操作会导致流之间的转化, 如filter(Predicate)一个终端操作: 终端操作会产生最终所需要的…

【考研英语】2011 年英语(一)排序题思路复盘(费曼学习法)

文章目录 引言一、找语段特征词二、确定位置写在最后 引言 英语一中的新题型之一 —— 排序题,我是看的刘琦老师的方法课,她用的 2011 年的真题来讲解方法。讲完让我们回去用“费曼学习法”复盘以下,我个人感觉是一个不错的方法,…

Altium Designer 批量添加元器件后缀

Altium Designer 批量添加元器件后缀 方法一方法二可能出现的问题要注意 方法一 您可以使用 Altium Designer 中的“批量修改元器件名称”功能来批量添加元器件后缀。具体步骤如下: 1.为了方便显示 操作流程,我这里复制了几个原理图的文件,粘…

8.3Jmeter使用json提取器提取数组值并循环(循环控制器)遍历使用

Jmeter使用json提取器提取数组值并循环遍历使用 响应返回值例如: {"code":0,"data":{"totalCount":11,"pageSize":100,"totalPage":1,"currPage":1,"list":[{"structuredId":&q…

“构建高效的前端表单验证与增删改功能实现“

目录 引言1. 基础增删改功能代码展示后端代码 2. 表单验证基本表单验证自定义验证规则 总结 引言 在现代Web开发中,前端表单验证和增删改功能是不可或缺的核心要素。本文将介绍如何使用ElementUI库来实现高效的前端表单验证和增删改功能。我们将从基础开始&#xf…

Flutter+SpringBoot实现ChatGPT流实输出

FlutterSpringBoot实现ChatGPT流式输出、上下文了连续对话 最终实现Flutter的流式输出上下文连续对话。 这里就是提供一个简单版的工具类和使用案例,此处页面仅参考。 服务端 这里直接封装提供工具类,修改自己的apiKey即可使用,支持连续…

FPGA project : TFT_LCD

实验目标: 驱动TFT_LCD显示十色彩条。 重点掌握的知识: 1,液晶显示器,简称LCD(Liquid Crystal Display),相对于上一代CRT显示器(阴极射线管显示器),LCD显示器具有功耗低、体积小、承载的信息量大及不伤眼…