数据倾斜
||约 8 分钟 · 2050 字|评论
请注意，本文编写于 876 天前，最后修改于 18 天前，其中某些信息可能已经过时。
常见的 Join
Common Join
一个 Join 就是一个 mapreduce 任务，但当多个 join 的字段间相同时，会使用一个 mr 任务就完成多个 join 任务。
sqlselect from t1 on t1.order_id = t2.order_id on t3.order_id = t2.order_id -- 此时就只会有一个mr任务！！！
Map Join
Hive 关于 mapjoin 的使用条件说明 在 map 端就完成了所有任务，避免了 shuffle，效率大大提升
- 适合场景
大表 Join 小表 - 原理
小表的数据在 join 前上传 hdfs，当运行任务时，从 hdfs 下载小表数据到 Maptask 的内存中，并逐行扫描大表的数据，实现 join - 使用
sql-- 开启map join set hive.auto.convert.join=true; -- 小表大小设置 --设置为 25MB set hive.mapjoin.smalltable.filesize=25000000; -- 无条件开启conditionaltask set hive.auto.convert.join.noconditionaltask=true; -- 设置conditionaltask的大小为10MB set hive.auto.convert.join.noconditionaltask.size=10000000;
Sorted Merge Join
- 原理
它的实现方式是将两个表按照连接键进行排序，然后进行合并。使用 Sorted Merge Join 可以减少 MapReduce 作业的数量，从而提高查询性能 - 使用
sqlSELECT /*+ MAPJOIN(a) MAPJOIN(b) */ * FROM table_a a JOIN table_b b ON a.key = b.key; -- 需要开启的参数 /* 1. hive.auto.convert.join：设置为 true，开启自动连接转换功能。 2. hive.optimize.sort.dynamic.partition：设置为 true，允许动态分区优化。 3. hive.optimize.sort.dynamic.partition.sortby：设置为 true，开启基于排序的动态分区优化。 4. hive.optimize.sortmerge.join：设置为 true，开启 Sorted Merge Join 优化。 5. hive.exec.parallel：设置为 true，开启并行执行。 */
- 适用场景
对 join 的 key 有顺序要求
Bucketed Map Join
适用于连接键基数较高的情况，这时候可以使用桶的元数据信息来直接跳过不匹配的行，从而提高查询性能。Bucket Map Join 在数据倾斜的情况下也表现良好，因为桶可以将数据均匀分布在不同的节点上，避免单节点负载过大的问题
- 原理
在 map 端按照分桶字段 Join，减少了较 Map Join 造成的小表全表加载，减轻每个 Join 压力，之后 reduce - 使用
sql--关闭cbo优化，cbo会导致hint信息被忽略 set hive.cbo.enable=false; --map join hint默认会被忽略(因为已经过时)，需将如下参数设置为false set hive.ignore.mapjoin.hint=false; --启用bucket map join优化功能 set hive.optimize.bucketmapjoin = true;
- 场景
大表 Join 大表
Sorted Merge Bucket Join
- 原理
相当于 Sorted Merge Join 和 Bucketed Map Join 的结合 - 使用
- Join 表要求分桶，且分桶数量必须为整数倍关系
- Join 的分桶表必须保证 分桶字段就是 Join 字段，并且也是分桶表的排序字段
sql--启动Sort Merge Bucket Map Join优化 hive.optimize.bucketmapjoin.sortedmerge=true; --使用自动转换SMB Join set hive.auto.convert.sortmerge.join=true;
- 场景
大表 Join 大表
数据倾斜
分组聚合
- 倾斜场景
sql-- 统计中国每个地区的人口数量 -- 会出现数据倾斜的sql select region,count(1) from population group by region;
- 倾斜原因
分组 key 分布不均匀造成，导致 reduce 端任务过大，执行缓慢或者失败，那么可以分两个方向思考- 数据源分布是否可改变？
- 是否可以对分组聚合的 key，做出调整，减轻数据倾斜
- 优化
- map 端的预聚合参数开启
**set hive.map.aggr=true;**
同时可以设置预聚合相关参数
- map 端的预聚合参数开启
sql-- 预聚合的条件 set hive.map.aggr.hash.min.reduction=0.5; --用于检测源表是否适合map-side聚合的条数。 set hive.groupby.mapaggr.checkinterval=100000; --map-side聚合所用的hash table，占用map task堆内存的最大比例，若超出该值，则会对hash table进行一次flush set hive.map.aggr.hash.force.flush.memory.threshold=0.9;
- 开启 Hive 提供的 GroupBy 倾斜优化参数
**set hive.groupby.skewindata=true;**
- 参数原理
hive.groupby.skewindata=true
对数据量大的 Join Key 单独做一次 Map Join，并和其他的 Join 结果做合并，此时会出现两个 MR 任务
hive.groupby.skewindata = true 参数无法开启吗？
手工实现预聚合
sql-- skewindata手动实现 select region,sum(1) from ( select region,count(1) from population group by region,cast(rand() * 7 as int) ) tmp group by region;
Join 形成的数据倾斜
- 场景
sqlselect od.order_detail_id, oi.order_id, oi.create_date, oi.order_id, total_amount from order_info oi join order_detail od on oi.order_id = od.order_id;
- 原因
Join Key 分布不均 - 优化
- Map Join 参数开启
**set hive.auto.convert.join=true;**，其他相关参数见官网 - Hive 提供的 Join 倾斜优化参数开启
**set hive.optimize.skewjoin=true;**，
自动对数据量大的 key 单独做 Mapjoin，默认 10 万个相同 key 进入一个 reduce ，进行二次聚合。其他相关参数见官网 - SMB Join
需要满足 SMB Join 的条件
- Map Join 参数开启
- 手动实现 SMB Join 效果
sql--- 使用扩容表方式join select od.order_detail_id, oi.order_id, oi.create_date, oi.order_id, total_amount from order_detail od -- 将较小的表order_info，增加字段new_id,并使用Union all 上下连接扩容后的表。右表随机且扩容 join (select order_id, concat(order_id, '-', 0) new_id, user_id, create_date, total_amount from order_info o union all select order_id, concat(order_id, '-', 1) new_id, user_id, create_date, total_amount from order_info o union all select order_id, concat(order_id, '-', 2) new_id, user_id, create_date, total_amount from order_info o ) oi -- 此时order_detail(大表)与小表order_info扩容后的新表的连接条件转化 -- 但需要保证通过扩容打散后join的最终结果一致，左表随机 on oi.new_id = concat(od.order_id, '-', `floor`(rand() * 3));
常见的数据倾斜的处理方式
测试数据：
sparksqlCREATE TABLE IF NOT EXISTS user_orders ( order_id STRING COMMENT '订单ID', user_id BIGINT COMMENT '用户ID', shop_account_id BIGINT COMMENT '商家店铺ID', order_amount DECIMAL(10,2) COMMENT '订单金额', order_status STRING COMMENT '订单状态', order_time STRING COMMENT '下单时间' ) COMMENT '用户订单事实测试表' PARTITIONED BY (dt STRING COMMENT '业务日期分区'); CREATE TABLE IF NOT EXISTS user_payments ( payment_id STRING COMMENT '支付流水ID', pay_order_id STRING COMMENT '支付关联业务单号', user_id BIGINT COMMENT '支付用户ID', shop_account_id BIGINT COMMENT '商家店铺ID', pay_amount DECIMAL(10,2) COMMENT '支付金额', pay_channel STRING COMMENT '支付渠道', pay_status STRING COMMENT '支付状态', pay_time STRING COMMENT '支付时间' ) COMMENT '用户支付事实测试表' PARTITIONED BY (dt STRING COMMENT '业务日期分区'); CREATE TABLE IF NOT EXISTS users ( user_id BIGINT COMMENT '用户ID', user_name STRING COMMENT '用户名', vip_level STRING COMMENT 'VIP等级' ) COMMENT '用户信息维度表'; insert overwrite table users select user_id, concat('user_', cast(user_id as string)) as user_name, case when user_id in (1,2,3,4,5,6,7,8,9,10) then 'VIP5' when user_id between 11 and 100 then 'VIP4' when user_id between 101 and 1000 then 'VIP3' when user_id between 1001 and 10000 then 'VIP2' else 'VIP1' end as vip_level from ( select explode(sequence(1, 100000)) as user_id ) t union all select 0 as user_id, 'user_0' as user_name, 'VIP0' as vip_level union all select -1 as user_id, 'user_-1' as user_name, 'VIP0' as vip_level union all select 999999999 as user_id, 'user_999999999' as user_name, 'VIP0' as vip_level; insert overwrite table user_orders partition (dt) select order_id, user_id, shop_account_id, order_amount, order_status, order_time, dt from ( -- 1) 普通背景订单 select concat('ord_norm_', cast(dt_id as string), '_', cast(user_id as string), '_', cast(seq_id as string)) as order_id, cast(user_id as bigint) as user_id, cast( case when pmod(user_id, 1000) = 0 then 0 when pmod(user_id, 777) = 0 then -1 when pmod(user_id, 666) = 0 then 999999999 when user_id between 201 and 500 then 10001 + pmod(user_id, 5) when user_id between 501 and 5000 then 10006 + pmod(user_id, 95) else 10101 + pmod(user_id, 9900) end as bigint ) as shop_account_id, cast((50 + pmod(user_id * 7 + seq_id, 1000)) / 10.0 as decimal(10,2)) as order_amount, case when pmod(user_id + seq_id, 10) = 0 then 'CANCEL' when pmod(user_id + seq_id, 7) = 0 then 'FINISHED' else 'CREATED' end as order_status, concat( case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end, ' ', lpad(cast(pmod(user_id + seq_id, 24) as string), 2, '0'), ':', lpad(cast(pmod(user_id * 3 + seq_id, 60) as string), 2, '0'), ':', lpad(cast(pmod(user_id * 7 + seq_id, 60) as string), 2, '0') ) as order_time, case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end as dt from ( select dt_id, user_id, seq_id from (select explode(array(1,2,3)) as dt_id) d cross join (select explode(sequence(201, 100000)) as user_id) u cross join (select explode(sequence(1, 3)) as seq_id) s ) t1 union all -- 2) 少量超热点用户订单 select concat('ord_superhot_', cast(dt_id as string), '_', cast(user_id as string), '_', cast(seq_id as string)) as order_id, cast(user_id as bigint) as user_id, cast(10001 + pmod(user_id, 5) as bigint) as shop_account_id, cast((100 + pmod(seq_id * 9, 800)) / 10.0 as decimal(10,2)) as order_amount, case when pmod(seq_id, 20) = 0 then 'CANCEL' when pmod(seq_id, 3) = 0 then 'FINISHED' else 'CREATED' end as order_status, concat( case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end, ' ', lpad(cast(pmod(seq_id, 24) as string), 2, '0'), ':', lpad(cast(pmod(seq_id * 2, 60) as string), 2, '0'), ':', lpad(cast(pmod(seq_id * 5, 60) as string), 2, '0') ) as order_time, case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end as dt from ( select dt_id, user_id, seq_id from (select explode(array(1,2,3)) as dt_id) d cross join (select explode(sequence(1, 10)) as user_id) u cross join (select explode(sequence(1, 20000)) as seq_id) s ) t2 union all -- 3) 中热点用户订单 select concat('ord_hot_', cast(dt_id as string), '_', cast(user_id as string), '_', cast(seq_id as string)) as order_id, cast(user_id as bigint) as user_id, cast(10006 + pmod(user_id, 95) as bigint) as shop_account_id, cast((80 + pmod(user_id + seq_id * 3, 600)) / 10.0 as decimal(10,2)) as order_amount, case when pmod(seq_id, 15) = 0 then 'CANCEL' when pmod(seq_id, 4) = 0 then 'FINISHED' else 'CREATED' end as order_status, concat( case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end, ' ', lpad(cast(pmod(user_id + seq_id, 24) as string), 2, '0'), ':', lpad(cast(pmod(user_id * 2 + seq_id, 60) as string), 2, '0'), ':', lpad(cast(pmod(user_id * 5 + seq_id, 60) as string), 2, '0') ) as order_time, case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end as dt from ( select dt_id, user_id, seq_id from (select explode(array(1,2,3)) as dt_id) d cross join (select explode(sequence(11, 200)) as user_id) u cross join (select explode(sequence(1, 1500)) as seq_id) s ) t3 union all -- 4) 异常用户 / 异常店铺订单 select concat('ord_invalid_', cast(dt_id as string), '_', cast(user_id as string), '_', cast(seq_id as string)) as order_id, cast(user_id as bigint) as user_id, cast(shop_account_id as bigint) as shop_account_id, cast((20 + pmod(seq_id * 4, 300)) / 10.0 as decimal(10,2)) as order_amount, 'CREATED' as order_status, concat( case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end, ' ', lpad(cast(pmod(seq_id, 24) as string), 2, '0'), ':', lpad(cast(pmod(seq_id * 2, 60) as string), 2, '0'), ':', lpad(cast(pmod(seq_id * 3, 60) as string), 2, '0') ) as order_time, case dt_id when 1 then '2025-01-01' when 2 then '2025-01-02' when 3 then '2025-01-03' end as dt from ( select dt_id, user_id, shop_account_id, seq_id from (select explode(array(1,2,3)) as dt_id) d cross join ( select cast(0 as bigint) as user_id, cast(0 as bigint) as shop_account_id union all select cast(-1 as bigint) as user_id, cast(-1 as bigint) as shop_account_id union all select cast(999999999 as bigint) as user_id, cast(999999999 as bigint) as shop_account_id ) u cross join (select explode(sequence(1, 25000)) as seq_id) s ) t4 ) all_orders; insert overwrite table user_payments partition (dt) select concat('pay_', order_id, '_', cast(pay_seq as string)) as payment_id, order_id as pay_order_id, user_id, shop_account_id, cast( case when pay_cnt = 1 then order_amount when pay_cnt = 2 and pay_seq = 1 then round(order_amount * 0.5, 2) when pay_cnt = 2 and pay_seq = 2 then order_amount - round(order_amount * 0.5, 2) when pay_cnt = 3 and pay_seq = 1 then round(order_amount * 0.4, 2) when pay_cnt = 3 and pay_seq = 2 then round(order_amount * 0.3, 2) else order_amount - round(order_amount * 0.4, 2) - round(order_amount * 0.3, 2) end as decimal(10,2) ) as pay_amount, case when pmod(abs(hash(concat(order_id, cast(pay_seq as string)))), 4) = 0 then 'ALIPAY' when pmod(abs(hash(concat(order_id, cast(pay_seq as string)))), 4) = 1 then 'WECHAT' when pmod(abs(hash(concat(order_id, cast(pay_seq as string)))), 4) = 2 then 'CARD' else 'BANK' end as pay_channel, case when pay_seq < pay_cnt then 'FAILED' else 'SUCCESS' end as pay_status, concat( substr(order_time, 1, 10), ' ', lpad(cast(pmod(abs(hash(concat(order_id, cast(pay_seq as string), '_hh'))), 24) as string), 2, '0'), ':', lpad(cast(pmod(abs(hash(concat(order_id, cast(pay_seq as string), '_mi'))), 60) as string), 2, '0'), ':', lpad(cast(pmod(abs(hash(concat(order_id, cast(pay_seq as string), '_ss'))), 60) as string), 2, '0') ) as pay_time, dt from ( select order_id, user_id, shop_account_id, order_amount, order_time, dt, pay_cnt from ( select order_id, user_id, shop_account_id, order_amount, order_time, dt, case when pmod(abs(hash(order_id)), 10) = 0 then 0 when pmod(abs(hash(order_id)), 20) = 0 then 3 when pmod(abs(hash(order_id)), 5) = 0 then 2 else 1 end as pay_cnt from user_orders where dt in ('2025-01-01', '2025-01-02', '2025-01-03') ) t where pay_cnt > 0 ) base lateral view explode(sequence(1, pay_cnt)) e as pay_seq;
数据特点： user_orders 表存在 user_id 为 null，999999999，小于 0 的无效记录，且这些特征的记录在与 users 关联时存在数据倾斜。
Join
无效 Join
- join key 如果不需要保留
使用 inner join 或者 直接过滤，user_id 为 null，999999999，小于 0 的记录
sparksqlselect count(*) from ( select t1.user_id, u.user_name, t1.order_time from user_orders t1 inner join users u on t1.user_id = u.user_id and t1.dt = '2025-01-03' ) ; select count(*) from ( select t1.user_id, u.user_name, t1.order_time from user_orders t1 left join users u on t1.user_id = u.user_id where t1.dt = '2025-01-03' and u.user_id is not null and u.user_id != 999999999 and u.user_id > 0 );
- join key 不需要保留
需要保留 user_id 为 null，999999999，小于 0 的记录，但是 user_id 为 null，999999999，小于 0 的记录 join 造成数据倾斜。
user_id 和 order_id 只有不小于 0 的才有效，并且 order_id 分布均匀
sparksqlselect count(*) from ( select t1.user_id, u.user_name, t1.order_time from user_orders t1 left join users u on if(u.user_id is null or u.user_id = 999999999 or u.user_id < 0, t1.order_id * -1, t1.user_id) = u.user_id where t1.dt = '2025-01-03' );
- 避免随机打散的值跟右表关联的上，比如右表也是正整数，那么左表随机打散不能使用正整数，避免碰撞上
- 如果表中没有分布较为均匀的 key，避免直接使用 rand()，因为 spark task 失败后重试会导致 rand(生成跟之前不一样的数据), 可以考虑使用 rand(3), rand(N)进行固定种子的随机
有效 Join
- 大表 JOIN 小表
设置小表广播的阈值，需要注意如果小表太大，反而广播可能会比 shuffle 更慢
sparksqlset spark.sql.autoBroadcastJoinThreshold = 100M; EXPLAIN EXTENDED select /*+MAPJOIN(T2)*/ t1.user_id, t2.user_name, t1.order_time from ( select user_id, order_time from user_orders where dt = '2025-01-03' ) t1 left join users t2 on t1.user_id = t2.user_id
- 大表 JOIN 大表
-
提高 shuffle 并发
这个参数决定 shuffle 之后，数据被分成多少份来并行处理，提高
spark.sql.shuffle.partitions的本质，不是“消灭数据倾斜”，而是把一次 shuffle 后的数据切得更细，让每个 reduce task 处理的数据更少，从而：- 提高并行度
- 降低单个 task 的数据量
- 缩短“非热点分区”的执行时间
- 在某些“轻度倾斜”场景下，缓解长尾任务问题
但是对 重度热点倾斜效果有限
sparksqlset spark.sql.shuffle.partitions = 400小表扩容，大表打散的方式
sparksqlselect t1.order_id, t1.order_time, t2.pay_time, t2.pay_amount from user_orders t1 left join user_payments t2 on t1.order_id = t2.pay_order_id and t2.dt = '2025-01-03' where t1.dt = '2025-01-03'; -- 优化 select t2.order_id, t2.order_time, t3.pay_time, t3.pay_amount from (select t1.order_id, t1.order_time, pmod(t1.shop_account_id, 10) pNum from mydb.user_orders t1 where t1.dt = '2025-01-03') t2 left join (select pay.pay_time, pay.pay_amount, pay.pay_order_id, seri.pNum from mydb.user_payments pay lateral view posexplode(split(space(10), ' ')) seri as pNum, val where pay.dt = '2025-01-03') t3 on t2.order_id = t3.pay_order_id and t2.pNum = t3.pNum;
其次如果可以定位 热key 的数据，也可以单独对热 key 的数据做大表打散，小表扩容的方式
Group by
两阶段聚合
适用于 count, sum, min, max, avg, 不适合直接用于 count(distinct)和一些不可分解的聚合
count 类型聚合拆解
sparksqlselect shop_account_id, count(*) order_num from mydb.user_orders t1 where dt = '2025-01-03' group by t1.shop_account_id; select shop_account_id, sum(order_date_num) order_num from ( select shop_account_id, count(*) order_date_num from mydb.user_orders t1 where dt = '2025-01-03' group by t1.shop_account_id, to_date(t1.order_time) ) tt group by shop_account_id;
avg 类型聚合拆解
sparksqlselect t1.pay_channel, avg(t1.pay_amount) avg_amount from mydb.user_payments t1 where dt = '2025-01-03' group by t1.pay_channel; -- 优化后 select pay_channel, sum(hour_amount) / sum(hour_times) from ( select t1.pay_channel, sum(t1.pay_amount) hour_amount, count(t1.pay_amount) hour_times from mydb.user_payments t1 where dt = '2025-01-03' group by t1.pay_channel,hour(t1.pay_time) ) t2 group by pay_channel
count(distinct)
待补充
row_number
也可使用两阶段
spark任务调优
map阶段（shuffle write）
这个阶段主要是数据的重分发阶段，产生中间文件，主要目标是为了减少一次性读取的数据数据量，减少shuffle过程中产生的中间文件
分区全量改增量
如果涉及的业务读取的历史全量分区中的部分数据，但是每次都需要扫描全量的分区，那么考虑将全量分区改造为增量分区，减少扫描的数据量
一级分区划分为多级分区
如果一级分区的划分规则下，每个分区的数据量依旧巨大，影响读取效率，可以改造为多级分区
使用distribute by 进行数据重排优化
按指定表达式重新分配数据，让后续任务按新的分区处理。它本身会产生 Shuffle；优化收益来自重新分配后，后续工作是否更均衡、数据分布是否更适合计算
reduce阶段（shuffle read）
此阶段为shuffle数据的拉取阶段
row_number的优化使用
row_number的使用对于涉及shuffle的过程，相较于使用group by 分组取极值的情况，没有数据的预聚合阶段
sparksql历史至今用户粒度末次交易的sku名称 历史至今用户粒度末次交易的交易金额等 源表order_info user_id tarns_tm trans_amt sku_nm 100 '2024-01-01 15:00:00' 1000 苹果 100 '2025-02-01 15:00:00' 1000 华为 需要结果 100 2000 华为 正常写法 select user_id ,sku_nm ,trans_amt from ( select user_id ,sku_nm ,trans_amt ,row_number() over(partition by user_id order by trans_tm desc) as rn from order_info where pt='${bizdate}' ) 优化写法 select user_id ,substr(max(concat(trans_tm,sku_nm)),20) as sku_nm ,substr(max(concat(trans_tm,trans_amt)),20) as trans_amt from order_info where pt='${bizdate}' group by user_id
数据回刷造成的小文件
生产环境中存在数据回刷的场景，比如对于存量数据的任务上线前，需要先处理存量数据，通常使用动态分区的特性处理；其次日常任务维护的过程中，如果出现故障需要数据的回刷，也称为补数据，也有可能使用动态分区的特性。 此时比如需要处理过去半年的数据，并通过动态分区回刷到对应的日期分区下
sparksqlset hive.exec.dynamic.partition.mode=nonstrict; insert overwrite table xxx.xxx partition(ds) select id ,xxx_date ,xxx_type ,pt from xxx.xxx where pt>='${bizdate}'
此时可能出现小文件问题，导致后续依赖该表的其他任务执行缓慢。出现小文件的原因：spark任务的产出文件数量取决于最后一个stage的task数（并行度） * 需要处理的分区数，如果有1000 个 Task×180 个分区=180,000 个小文件！下游任务读取这 18 万个小文件时，大量的 Split 和 NameNode RPC 交互直接导致下游任务运行 卡死。
通用解决方式：对已经存在的小文件借助spark3的AQE功能进行数据回刷
sparksql以下以 Spark 3.5 为例，展示写入独立中间表的思路： SET spark.sql.adaptive.enabled=true; SET spark.sql.adaptive.coalescePartitions.enabled=true; -- 优先参考目标大小，而不是尽量保留并行度 SET spark.sql.adaptive.coalescePartitions.parallelismFirst=false; SET spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB; INSERT OVERWRITE TABLE audit_log_compacted PARTITION(pt) SELECT /*+ REBALANCE(pt) */ id, xxx_date, xxx_type, pt FROM audit_log WHERE pt BETWEEN '${start_pt}' AND '${end_pt}';
原理说明：决定小文件数量的是公式最后一个stage的task数（并行度） * 需要处理的表分区数，需要处理的分区数量是不变的，因此从spark的task的数量入手，reduce task（shuffle read）的数量和需要处理的分区数量是直接关联的，使用AQE可以动态分析shuffle write后的分区数据量，动态合并小分区，从而减少下游reduce task的数量。REBALANCE(pt) 会尽量按日期组织数据，并结合 AQE 调整过小或过大的执行分区，它专门适用于希望控制写出文件大小的场景，依赖 AQE 开启。官方说明
plaintextStage 1 (Map 阶段) │ ▼ Shuffle Write（产生大量小 Partition，如每个只有几 MB） │ 【AQE 介入监控】统计每个 Shuffle Partition 的实际字节大小 │ ▼ 动态合并（Coalesce Shuffle Partitions） Stage 2 (Reduce 阶段写出) Task 数量被自动从 2000 压至合理的几十/上百个 │ ▼ 输出文件大小稳定在 128MB / 256MB 左右
版权声明