SPARK Expand问题的解决(由count distinct、group sets、cube、rollup引起的)

简介: SPARK Expand问题的解决(由count distinct、group sets、cube、rollup引起的)

背景


本文基于spark 3.1.2

我们知道spark对于count(distinct)/group sets 以及cube、rollup的处理都是采用转换为Expand的方法处理,

这样做的优点就是在数据量小的情况下,能有以空间换时间,从而达到加速的目的。

但是弊端也是很明显,就是在数据量较大的情况下,而且expand的倍数达到上百倍或者千倍的时候,这任务运行的时间很长(这在数分中是非常常见的)。


分析


先来看一组图:

image.png

是不是很刺激,数据从2,635,978,109直接扩张到了168,702,598,976,将近80倍。

该sql就是简单的读取表让后group by cube,如下:

image.png

该sql运行的时长达到了5个小时,如下:

image.png

经过优化后,该sql只需要49分钟,如下:

image.png

其实解决方法很简单,因为我们读取的是parquet的文件,且依赖的表的文件个数有400个,但是优化前的任务数是99个,所以我们可以设置spark.sql.files.maxPartitionBytes的值来控制每个task任务读取的数据大小,笔者是设置为20MB。具体spark是怎么读取parquet文件的可以参考Spark-读取Parquet-为什么task数量会多于Row Group的数量。


结论


这种expand问题解决的思路也是有的:


设置spark.sql.files.maxPartitionBytes为合适的值,这种只适合直接依赖于表的情况(不适用子查询)

参考SPARK-32542,这种只适合group sets的情况,有可能会导致ExchangeExec过多的问题

repartition 中间结果表,再拿中间临时结果作为依赖表,这种如果依赖的表很多,需要建立很多的临时表,比较繁琐

 create table temp_a select /*+ repartition(1000) */ from fackt_table
 select columns from temp_a group by cube()
  1. 修改spark源码从源码底层支持(后续文章会说到)
相关文章
|
5月前
|
存储 分布式计算 Apache
Spark编程范例:Word Count示例解析
Spark编程范例:Word Count示例解析
|
SQL 分布式计算 Spark
如何在Spark中实现Count Distinct重聚合
Count Distinct是SQL查询中经常使用的聚合统计方式,用于计算非重复结果的数目。由于需要去除重复结果,Count Distinct的计算通常非常耗时。本文主要介绍在Spark中如何基于重聚合实现交互式响应的COUNT DISTINCT支持。
|
SQL 分布式计算 Spark
SPARK SQL中 Grouping sets转Expand怎么实现的(逻辑计划级别)
SPARK SQL中 Grouping sets转Expand怎么实现的(逻辑计划级别)
540 0
|
SQL 分布式计算 自然语言处理
Spark 系列教程(1)Word Count
Spark 系列教程(1)Word Count
449 0
Spark 系列教程(1)Word Count
|
SQL 存储 分布式计算
如何在Spark中实现Count Distinct重聚合
背景 Count Distinct是SQL查询中经常使用的聚合统计方式,用于计算非重复结果的数目。由于需要去除重复结果,Count Distinct的计算通常非常耗时。为了支持更快速的非重复结果统计Spark还基于Hyperloglog实现了Approximate Count Distinct,用于统计非重复结果的近似值,支持。
3804 0
|
分布式计算 API Spark
Spark Streaming和Flink的Word Count对比
准备: nccat for windows/linux 都可以 通过 TCP 套接字连接,从流数据中创建了一个 Spark DStream/ Flink DataSream, 然后进行处理, 时间窗口大小为10s 因为 示例需要, 所以 需要下载一个netcat, 来构造流的输入。
2420 0
|
分布式计算 Spark
Spark sc.textFile(...).map(...).count() 执行完整流程
本文介绍下Spark 到底是如何运行sc.TextFile(...).map(....).count() 这种代码的,从driver端到executor端。
4192 0
|
13天前
|
存储 分布式计算 算法
大数据-106 Spark Graph X 计算学习 案例:1图的基本计算、2连通图算法、3寻找相同的用户
大数据-106 Spark Graph X 计算学习 案例:1图的基本计算、2连通图算法、3寻找相同的用户
35 0
|
13天前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
29 0
|
13天前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
40 0