Spark Shuffle 分区原理-北京尧图网络

一、Shuffle 分区基础原理1. 分区控制参数Map阶段并行度总核 * 2spark.default.parallelismcore_nums * 2shuffle write /shuffle read 阶段分区数由 spark.sql.shuffle.partitions 控制Spark 2.x/3.x 默认 200大数据量场景严重不足。2. Shuffle 触发判断看 Stage 是否存在宽依赖一个父分区被多个子分区复用触发 Shufflejoin / groupBy / aggregate / distinct / repartition / 窗口函数不触发 Shufflemap / filter / select / withColumn 窄依赖物理计划出现 Exchange / ShuffleExchange 节点 → 必然触发 Shuffle3. 分区数计算经验公式推荐分区数 ≈ Shuffle 总数据量 ÷ 目标分区大小目标分区大小256MB~512MB平衡内存 OOM 与 HDFS 小文件下限不低于集群总核心数保证基础并行度上限不超过总数据量 ÷ 64MB防止小文件爆炸4. 分区数调优利弊增大分区数优点单分区数据量变小、规避 Executor OOM、提升并行度缺点Task 和中间文件变多、加重调度与 HDFS I/O 压力二、Spark 3.x AQE配置自动自适应并行度、自动合并小分区、自动倾斜 Join 优化、根治小文件与 OOM。# 核心总开关 AQE 自适应全开 spark.sql.adaptive.enabled true; spark.sql.adaptive.logLevel WARN; # 1. 分区合并解决小Task、合并输出小文件 spark.sql.adaptive.coalescePartitions.enabled true; # 计算阶段小于64MB就合并只管计算不管最终文件基准阈值别动 spark.sql.adaptive.coalescePartitions.minPartitionSize 67108864; # 计算用足够大保证 100GB 数据计算飞快 spark.sql.shuffle.partitions 200; -- 保持默认不要改小 # 写入用AQE 自动合并小文件目标输出文件/分区理想大小 256MB spark.sql.adaptive.advisoryPartitionSizeInBytes 268435456; # AQE 合并 shuffle 分区时强制保留的「最少分区数」防止合并太狠、只剩 1~2 个大分区。 spark.sql.adaptive.coalescePartitions.minPartitionNum 1; # 【关键】写入时强制合并彻底解决小文件 spark.sql.adaptive.forceCoalesceOnInsertion.enabled true; # 开启 AQE 自适应写文件最优雅不用写 coalesce 代码 # 无需手动合并分区写入时自动把小分区合并成目标大小文件参数 spark.sql.adaptive.writeEnabledtrue spark.sql.adaptive.writeTargetFileSize128m # 2. 自动数据倾斜 Join 优化 # 自动识别并打散 Join 倾斜无需手动 salting spark.sql.adaptive.skewJoin.enabled true; # 单分区超过256MB 判定为倾斜分区 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes 268435456; # 超过平均分区5倍判定倾斜默认5不用改 spark.sql.adaptive.skewedPartitionFactor 5; # 3. 增强Shuffle重平衡均衡分区、固化文件大小 # 本地读 Shuffle减少网络 IO提速明显。 spark.sql.adaptive.localShuffleReader.enabled true; # Shuffle 后自动重新均衡分区大小。 spark.sql.adaptive.rebalancePartitions.enabled true; # 重平衡目标分区256MB重平衡目标分区大小和落地文件对齐不乱飘。 spark.sql.adaptive.rebalancePartitionSizeInBytes 268435456; # 4. 额外兜底防小文件、读写对齐 # 读文件按 128MB 切片和 HDFS 默认块对齐读性能最优。 spark.sql.files.maxPartitionBytes 134217728; # 控制一次任务最多能生成多少个动态分区避免动态分区产生大量小文件动态分区必加避免无限创建分区 spark.sql.dynamicPartitioning.maxNumDynamicPartitions 10000;简易配置规则数据量 50GB开启前 3 项基础 AQE 即可50GB1TBspark.sql.shuffle.partitions 设 1000~20001TB4TBspark.sql.shuffle.partitions 设 2000~4000最终落地文件统一控制在 128~256MB三、写入重分区核心说明Shuffle 聚合后数据量通常大幅压缩若不做写入重分区极易生成大量小文件拖垮 HDFS 元数据和查询性能。四、重分区三种实战方案方法一PySpark DataFrame API 方案核心要点repartition(n, 列名)触发 Shuffle数据均匀打散推荐用于写入前重分区coalesce(n)无 Shuffle仅能减少分区、数据分布不均生产谨慎使用写入动态分区必须配置 Hive 动态分区参数maxRecordsPerFile 限制单文件行数避免超大文件frompyspark.sqlimportSparkSessionfrompyspark.sql.functionsimport*frompyspark.sql.windowimportWindow# 1. 初始化SparkSession 生产标准配置sparkSparkSession.builder \.appName(Ecommerce_Order_Analysis)\.config(spark.sql.shuffle.partitions,2000)\.config(spark.sql.adaptive.enabled,true)\.config(spark.sql.adaptive.coalescePartitions.enabled,true)\.config(spark.sql.adaptive.advisoryPartitionSizeInBytes,256000000)\.config(hive.exec.dynamic.partition,true)\.config(hive.exec.dynamic.partition.mode,nonstrict)\.config(hive.exec.max.dynamic.partitions,20000)\.enableHiveSupport()\.getOrCreate()# 2. 数据源读取orders_dfspark.read.parquet(/data/orders)users_dfspark.read.parquet(/data/users)products_dfspark.read.parquet(/data/products)# 3. 业务计算含多轮Shuffleenriched_ordersorders_df \.join(broadcast(users_df),user_id,left)\.join(products_df,product_id,left)\.groupBy(category,region,dt)\.agg(sum(amount).alias(daily_sales),countDistinct(user_id).alias(unique_customers),avg(price).alias(avg_price))# 窗口函数触发Shufflewindow_specWindow.partitionBy(dt).orderBy(col(daily_sales).desc())enriched_ordersenriched_orders.withColumn(sales_rank,rank().over(window_spec))# 4. 写入前重分区推荐用法final_dfenriched_orders \.repartition(300,dt,region)\.sortWithinPartitions(category)# 不推荐coalesce 数据不均匀易倾斜# final_df enriched_orders.coalesce(100)# 5. 写入Hive分区表final_df.write \.mode(overwrite)\.partitionBy(dt,region)\.option(path,/user/hive/warehouse/sales.db/daily_summary)\.option(maxRecordsPerFile,2000000)\.option(compression,snappy)\.saveAsTable(sales.daily_summary)工具函数动态计算最优分区数defget_ideal_partitions(df,target_mb256,min_p20,max_p1000):按数据量自动计算最优分区数df.cache()df_size_mbdf.rdd.map(lambdax:len(str(x))).sum()/(1024*1024)idealmax(min_p,min(max_p,int(df_size_mb/target_mb)))print(f数据量{df_size_mb:.2f}MB最优分区数{ideal})returnideal# 调用示例ideal_partget_ideal_partitions(enriched_orders)final_dfenriched_orders.repartition(ideal_part,dt,region)数据倾斜手动处理工具函数defhandle_skewness(df,skew_key,total_partitions200):倾斜键加随机前缀打散适用于AQE无法自动优化场景frompyspark.sql.functionsimportconcat,lit,rand skewed_keys[category_A,category_B]df_skeweddf.filter(col(skew_key).isin(skewed_keys))df_normaldf.filter(~col(skew_key).isin(skewed_keys))df_skewed_processeddf_skewed \.withColumn(skew_key_with_prefix,concat(lit(prefix_),(rand()*10).cast(int).cast(string),lit(_),col(skew_key)))df_skewed_repartitioneddf_skewed_processed.repartition(total_partitions,skew_key_with_prefix)df_normal_repartitioneddf_normal.repartition(total_partitions,skew_key)returndf_skewed_repartitioned.union(df_normal_repartitioned)方法二Spark SQL 重分区方案核心关键字DISTRIBUTE BY按字段重分区 ShuffleDISTRIBUTE BY cast(rand() * n as int)重分区 Shuffle成n个SORT BY分区内局部排序CLUSTER BY等价 DISTRIBUTE BY SORT BY/* REPARTITION(n, 列) */Hint 强制指定分区数-- 全局参数设置SETspark.sql.shuffle.partitions2000;SETspark.sql.adaptive.enabledtrue;SEThive.exec.dynamic.partitiontrue;SEThive.exec.dynamic.partition.modenonstrict;SEThive.exec.max.dynamic.partitions10000;SEThive.exec.max.dynamic.partitions.pernode1000;-- 注册临时视图CREATEORREPLACETEMPORARYVIEWorders_viewASSELECT*FROMparquet./data/orders;CREATEORREPLACETEMPORARYVIEWusers_viewASSELECT*FROMparquet./data/users;CREATEORREPLACETEMPORARYVIEWproducts_viewASSELECT*FROMparquet./data/products;-- 业务聚合计算CREATEORREPLACETEMPORARYVIEWenriched_orders_viewASWITHjoined_dataAS(SELECTo.*,u.user_name,u.user_level,p.category,p.product_name,p.priceFROMorders_view oLEFTJOINusers_view uONo.user_idu.user_idLEFTJOINproducts_view pONo.product_idp.product_id),aggregatedAS(SELECTcategory,region,dt,SUM(amount)ASdaily_sales,COUNT(DISTINCTuser_id)ASunique_customers,AVG(price)ASavg_priceFROMjoined_dataGROUPBYcategory,region,dt)SELECT*,RANK()OVER(PARTITIONBYdtORDERBYdaily_salesDESC)ASsales_rankFROMaggregated;-- 写入重分区方式1DISTRIBUTE BY SORT BYINSERTOVERWRITETABLEsales.daily_summaryPARTITION(dt,region)SELECTcategory,daily_sales,unique_customers,avg_price,sales_rank,dt,regionFROMenriched_orders_view DISTRIBUTEBYdt,region SORTBYcategory;-- 写入重分区方式2Hint强制分区INSERTOVERWRITETABLEsales.daily_summaryPARTITION(dt,region)SELECT/* REPARTITION(300, dt, region) */category,daily_sales,unique_customers,avg_price,sales_rank,dt,regionFROMenriched_orders_view;-- 写入重分区方式3CLUSTER BY 等价分区排序INSERTOVERWRITETABLEsales.daily_summaryPARTITION(dt,region)SELECTcategory,daily_sales,unique_customers,avg_price,sales_rank,dt,regionFROMenriched_orders_view DISTRIBUTEBYdt,region CLUSTERBYcategory;分桶表创建与写入-- 创建分桶表CREATETABLEIFNOTEXISTSsales.bucketed_summary(category STRING,daily_salesDOUBLE,unique_customersBIGINT,avg_priceDOUBLE,sales_rankINT)PARTITIONEDBY(dt STRING,region STRING)CLUSTEREDBY(category)INTO50BUCKETS STOREDASORC TBLPROPERTIES(orc.compressSNAPPY,transactionaltrue);-- 写入分桶表INSERTOVERWRITETABLEsales.bucketed_summaryPARTITION(dt,region)SELECTcategory,daily_sales,unique_customers,avg_price,sales_rank,dt,regionFROMenriched_orders_view;方法三动态分区分桶写入优化# 动态分区多级分区重分区控制小文件transformedDF \.repartition(200,col(dt),col(category)).write.mode(overwrite).partitionBy(dt,category).option(maxRecordsPerFile,1000000).option(compression,snappy).saveAsTable(target_table)# 分桶表写入适合高频关联查询场景transformedDF \.write.mode(overwrite).bucketBy(50,category).sortBy(category).saveAsTable(bucketed_table)五、监控与验证 SQL / 代码查看执行计划final_df.explain(True)表统计信息收集ANALYZETABLEsales.daily_summaryCOMPUTESTATISTICS;核查各分区文件数量SELECTdt,region,COUNT(DISTINCTINPUT__FILE__NAME())asfile_countFROMsales.daily_summaryGROUPBYdt,regionORDERBYfile_countDESC;数据倾斜分区检测SELECTdt,region,COUNT(*)asrow_count,AVG(daily_sales)asavg_salesFROMsales.daily_summaryGROUPBYdt,regionHAVINGCOUNT(*)1000000ORDERBYrow_countDESCLIMIT10;广播 Join 优化 HintSELECT/* MAPJOIN(small_table) */*FROMlarge_tableJOINsmall_tableONlarge_table.keysmall_table.key;六、注意事项禁止滥用 coalesce无 Shuffle 合并数据分布不均极易引发数据倾斜写入前务必按Hive 分区字段做 repartition是根治小文件最有效手段必须配置 maxRecordsPerFile避免单文件过大影响查询效率Spark3.x 强制开启 AQE自动合并分区、自动倾斜优化减少人工调参动态分区作业必须配置 Hive 动态分区参数否则报错或分区生成异常小文件标准治理方案AQE开启写入前repartition分区列合理文件大小控制小表优先使用 broadcast 广播 Join直接消除 Shuffle 开销分桶表适合大表高频 Join、聚合场景可大幅减少查询 Shuffle避免过度分区太多小文件会影响 HDFS 和查询性能考虑存储格式ORC/Parquet 有最小文件大小要求监控资源使用观察 Executor 内存和 GC 情况结合业务特点根据数据倾斜情况调整分区策略七、调优口诀Shuffle 并行度50GB 内 10001TB 内 20004TB 左右 4000AQE 三件套全开自适应分区合并倾斜优化写入必做repartition 按分区字段重分区文件大小严控统一 128~256MB小文件根源未做写入前重分区 AQE 未开启

Spark Shuffle 分区原理

相关推荐

C SV--数据类型对应关系及内存映射

第7课：TensorFlow｜TF自动梯度原理【反向传播底层逻辑、梯度计算、梯度裁剪实操】

Excel筛选全攻略：从基础操作到Python自动化实战

更多资讯

30天煤气特种作业证K23实操不挂科的内部消息，老电工都这么干

天津建筑电工证培训通过率揭秘：学历不高怕报不上名？别慌

在深圳哪里办理电工证怎么报名？怕考不过白交培训费？教你一招保过

唐山特种工考证官方回应：怕考不过白交培训费？别急，这篇全搞定

唐山特种工考证过期了怎么复审？考前押题帮你搞定

国网的电工证能干嘛怎么考

今日更新

金坛办电工证在哪里？考前押题能帮你少走弯路

怎样查看真假电工证怎么考

史密斯内部电工证值不值得考？怕考不过白交培训费的看过来

本周精选

安监局电工证分数查询怎么查才能知道含金量高不高

新疆电工作业干活累吗良心建议助你避开考不过的坑

中牟县电工证怎么报名外省证能转吗

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

读完文章，想聊聊您的网站？