1. 项目概述为什么“Spark算子”不是个技术名词而是一把数据仓库工程师的瑞士军刀在某高校数据平台组做支撑的那两年我每天打开监控看板的第一件事不是查集群负载而是扫一眼最近提交的Spark作业里reduceByKey和groupByKey的调用比例——这个比值一旦超过1.8基本就能预判今天下午三点会有三四个业务方来工位前蹲点手里攥着刚被OOM kill掉的作业日志眼神里写满“求救”。这不是玄学是Spark算子在真实数据仓库场景中不可替代的实操地位决定的它不单是API列表里的函数名而是数据倾斜治理、内存压测边界、ETL链路吞吐量设计的具象化落点。你翻遍所有标榜“精通Spark”的简历真正能说清mapPartitions比map省多少序列化开销、repartition和coalesce在宽依赖场景下触发Shuffle的物理差异、甚至foreachPartition里连接池复用失败导致连接数爆炸的根因不超过两成。这篇内容就专为那些已经写过五十个以上Spark SQL但还没亲手调过spark.sql.adaptive.enabled开关的人准备——我们不讲RDD API文档只拆解你在构建T1用户行为宽表、实时数仓分层聚合、跨源维度关联时每一个算子选择背后的真实代价、可验证的阈值、以及被官方文档悄悄省略的“但书”。核心关键词全部落在“数据仓库”和“Spark算子”上这意味着所有讨论都锚定在离线批处理、高并发调度、TB级中间结果落地、UDF稳定性保障这些具体战场而不是泛泛而谈“分布式计算原理”。2. Spark算子在数据仓库中的角色定位与设计逻辑2.1 数据仓库场景对Spark算子的特殊要求不是快而是稳、准、可预期很多刚从Flink实时开发转岗来的同学第一次写Spark离线任务时会本能地套用keyBy → process的思维结果在调度平台上看到作业运行时间从15分钟跳到3小时还查不出原因。根本差异在于实时计算追求低延迟可以接受少量乱序或状态抖动而数据仓库的SLA是“T1凌晨两点前必须产出全量报表”它要求的是确定性。这种确定性直接映射到算子设计上形成三个硬约束内存可控性groupByKey会把同一key的所有value拉到单个Executor内存里当用户标签表里出现“未知兴趣”这类高频key时极易触发GC风暴。而reduceByKey在Map端就做了局部聚合内存占用下降60%以上实测某电商用户画像job从平均4.2GB/Executor压到1.6GB。这不是理论值是我们在YARN队列资源配额卡死时唯一能快速止损的手段。Shuffle可预测性数据仓库ETL链路里90%的性能瓶颈发生在Shuffle阶段。repartition(200)看似简单但若上游分区数只有4Spark会强制触发全量Shuffle而coalesce(200)在减少分区时不会Shuffle但增加分区时根本无效。我们曾为一个日志清洗任务反复调整分区数最终发现最优解是repartition(192)——因为集群总核数是192每个Task恰好绑定一个CPU核心避免了线程争抢。这个数字不是拍脑袋是通过spark.sql.adaptive.coalescePartitions.enabledtrue开启自适应后观察AdaptiveSparkPlan日志里CoalescedPartition的实际合并效果反推出来的。血缘可追溯性cache()和persist(StorageLevel.MEMORY_AND_DISK)看着只是缓存策略差异但在数据仓库的血缘追踪系统里前者会让下游所有依赖该RDD的作业在元数据里标记为“缓存失效风险高”。某次大促期间一个被cache()的订单汇总表因Executor宕机丢失导致下游17个报表任务全部重跑。后来我们强制推行persist(StorageLevel.MEMORY_AND_DISK_SER)序列化存储磁盘兜底虽然首次读取慢8%但血缘系统能准确识别“该RDD已持久化且具备容错能力”调度器自动跳过重算。提示数据仓库场景下算子选择的第一判断标准永远不是“功能是否满足”而是“该操作是否引入不可控的内存/网络/磁盘变量”。任何需要依赖“运气”比如shuffle partition数刚好匹配数据分布的方案都应该被标记为高危。2.2 算子分类的本质按数据流动路径而非API形态划分官方文档把算子分成Transformation和Action但这对数据仓库工程师毫无指导意义。我们按数据在ETL管道中的实际流向重新划分为三类分流型算子filter、sample、randomSplit。它们不改变数据结构只做行级筛选。关键陷阱在于filter的谓词下推——当你的SQL里写WHERE dt20240101 AND user_id 1000000Spark Catalyst会自动把分区过滤dt下推到Scan阶段但user_id条件必须等数据读入内存后才执行。这意味着即使你只查一天数据仍要扫描整个分区文件。解决方案是配合Hive表的分桶字段把user_id设为分桶键让filter能利用分桶索引跳过无关文件块。聚合型算子reduceByKey、aggregateByKey、foldByKey。它们的核心差异在初始值处理逻辑。reduceByKey要求函数满足结合律和交换律如而aggregateByKey允许你定义独立的zeroValue和seqOp/combOp适合计算去重UV这类场景。某次我们统计DAU用reduceByKey((x,y) xy)得到错误结果——因为原始数据里有重复曝光记录正确做法是aggregateByKey(zeroValueSet.empty)(seqOp(set,log)setlog.user_id, combOp(__))再取size。这里zeroValue必须是可变集合否则combOp合并时会创建新对象内存暴涨。重组型算子join、cogroup、leftOuterJoin。数据仓库最常踩坑的领域。join默认使用SortMergeJoin当小表10MB时Catalyst会自动转为BroadcastHashJoin。但这个阈值是静态配置的spark.sql.autoBroadcastJoinThreshold10485760如果小表实际大小是10.2MB就会退化成Shuffle Join。我们曾为一个维度表城市编码→城市名手动添加broadcast()提示df.join(broadcast(dim_city), city_id)把join耗时从8分钟压到42秒。注意broadcast()不是万能的当维度表更新频繁时广播变量会随每次作业提交重新分发反而增加Driver压力。2.3 为什么数据仓库必须关注算子组合的“副作用链”单个算子再规范组合起来也可能产生灾难性后果。我们曾遇到一个经典案例某用户行为宽表任务逻辑是“先filter出有效事件再map解析JSON最后reduceByKey聚合”。表面看无懈可击但上线后发现每天凌晨三点集群IO打满。根源在于map解析JSON时Spark默认使用UnsafeRow格式而JSON解析库Jackson返回的是JavaHashMap对象导致每行数据都要经历“UnsafeRow → Java Object → UnsafeRow”的反复序列化。解决方案是改用mapPartitions在分区粒度上复用JacksonObjectMapper实例并用KryoSerializer替代默认序列化器。这个改动让序列化耗时下降73%IO等待减少40%。更隐蔽的副作用是cache()位置的选择。常见错误是在filter后立即cache()但filter后的数据分布可能极不均匀比如过滤掉99%的测试流量剩下1%全是真实用户而用户ID集中在几个号段。此时cache()会把热点数据全塞进少数Executor造成负载倾斜。正确姿势是filter → repartition → cache()用repartition(200)强制打散数据再缓存。虽然多了一次Shuffle但后续所有依赖该缓存的作业都能获得均衡负载。注意数据仓库的算子链不是线性流水线而是带反馈的闭环系统。每个cache()点都是潜在的故障放大器每个repartition()都是对数据分布的主动干预。必须用explain(true)反复查看物理执行计划确认Shuffle Read/Write的字节数是否符合预期。3. 核心算子深度解析与仓库级实操要点3.1reduceByKey数据仓库ETL的基石但90%的人用错了初始值reduceByKey号称“Spark最高效聚合算子”但它的高效完全建立在正确使用基础上。问题出在zeroValue的设计上——很多人直接写reduceByKey(_ _)以为加法天然安全。实际上当value是字符串拼接时如reduceByKey(_ , _)空字符串作为zeroValue会导致首尾多出冗余逗号当value是List时如reduceByKey(_ _)Nil作为zeroValue会在合并时创建新List对象引发GC风暴。真实仓库场景案例某广告归因任务需将用户点击流按session_id聚合为有序事件序列。原始代码val sessionEvents logs .map(log (log.session_id, List(log))) .reduceByKey(_ _)运行三天后Executor OOM频发。根因是List(log)每次创建新对象操作符生成新List内存无法及时回收。改造方案// 使用mutable.ArrayBuffer复用内存 val sessionEvents logs .map(log (log.session_id, ArrayBuffer(log))) .reduceByKey((buf1, buf2) buf1 buf2) // 是原地追加 .mapValues(_.toList) // 最后转成不可变List供下游使用关键点在于操作符它直接在原有ArrayBuffer上追加避免了对象创建。实测内存峰值下降58%GC时间从12s/分钟降到1.3s/分钟。另一个致命误区是忽略reduceByKey的Map端聚合机制。当你的seqOp函数如计算成本很高时比如涉及正则匹配Map端聚合会重复执行多次。某日志清洗任务中reduceByKey的value是经过parseUserAgent(userAgent)解析后的设备信息而parseUserAgent内部调用正则引擎耗时20ms。由于Map端聚合会为每个key的每个value调用一次seqOp导致单个Task CPU使用率长期95%以上。解决方案是改用aggregateByKey把解析逻辑移到seqOp外部val parsedLogs logs.map { log val device parseUserAgent(log.userAgent) // 一次解析 (log.session_id, (device, log.event_time)) } val aggregated parsedLogs .aggregateByKey(zeroValue (Device.empty, 0L))( seqOp (acc, curr) (acc._1.merge(curr._1), math.max(acc._2, curr._2)), combOp (acc1, acc2) (acc1._1.merge(acc2._1), math.max(acc1._2, acc2._2)) )这样parseUserAgent只执行一次聚合逻辑轻量化CPU使用率回落至65%以下。3.2join数据仓库最昂贵的操作如何用Hint和配置把它变成“廉价”的在数据仓库join不是功能实现手段而是资源消耗计量单位。我们内部有个经验值一个未优化的join操作其Shuffle数据量通常是参与表原始大小的3~5倍。原因在于SortMergeJoin需要对两边数据按join key全量排序而排序过程会产生大量中间文件。BroadcastJoin的实战边界spark.sql.autoBroadcastJoinThreshold10MB是默认值但生产环境我们强制设为5MB。为什么因为广播变量会加载到每个Executor的JVM堆内存当Executor配置为8G内存时10MB广播变量占堆内存0.125%看似安全。但当同时运行20个作业每个都广播一个10MB表瞬间吃掉200MB堆内存触发Full GC。我们通过监控jvm.heap.used指标发现GC频率与广播表大小呈强正相关最终将阈值下调至5MB并配套建立维度表大小巡检机制——任何超过5MB的维度表必须走repartitionsort预处理或拆分为高频/低频两个子表。SortMergeJoin的调优密钥当必须走Shuffle Join时核心是控制spark.sql.adaptive.enabledtrue开启后的自适应行为。重点参数spark.sql.adaptive.localShuffleReader.enabledtrue允许本地读取Shuffle文件减少网络传输。某次测试显示开启后Shuffle Read Time下降35%。spark.sql.adaptive.coalescePartitions.enabledtrue动态合并小分区。但要注意它只对Shuffle后的小分区生效对repartition产生的分区无效。我们曾误以为开启此参数就能解决小文件问题结果发现repartition(1000)产生的1000个分区即使每个只有1MB也不会被合并——因为这是显式重分区不属于Shuffle阶段的自适应范围。最危险的join类型leftOuterJoin。它要求左表全量保留右表缺失时填充null。但当右表存在数据倾斜比如某个城市ID在维度表里对应10万条记录而其他城市平均只有100条leftOuterJoin会把左表所有匹配该城市的记录全部Shuffle到同一个Task。解决方案是“倾斜Key分离”先用filter提取右表的倾斜key如dim_city.filter($city_id 100000)对左表做leftOuterJoin时用broadcast加载倾斜key对应的右表子集对剩余非倾斜key走正常SortMergeJoin最后union两个结果集这个方案把原本需要30分钟的join压缩到4分12秒且资源消耗稳定。3.3mapPartitions被严重低估的“性能核弹”但用错就是内存炸弹mapPartitions的价值在于规避map的逐行序列化开销但它也是数据仓库最易误用的算子之一。典型错误是把它当成map的“加强版”直接在里面写业务逻辑结果OOM频发。正确用法一连接池复用。数据仓库常需在转换中查维表比如根据用户ID查会员等级。若用map每个record都会新建数据库连接// 错误示范每行创建新连接 df.map(row { val conn DriverManager.getConnection(url) // 每行都new connection val level queryLevel(conn, row.userId) Row.merge(row, Row(level)) })改用mapPartitions在分区粒度复用连接// 正确示范分区级连接复用 df.mapPartitions { iter val conn DriverManager.getConnection(url) val stmt conn.prepareStatement(SELECT level FROM dim_user WHERE user_id ?) iter.map { row stmt.setLong(1, row.getLong(0)) val rs stmt.executeQuery() val level if (rs.next()) rs.getInt(level) else 0 Row.merge(row, Row(level)) }.buffered // 加buffered避免iterator提前关闭conn }注意buffered调用它确保iterator被消费完后再释放连接。实测连接创建耗时从12ms/行降至0.3ms/行整体耗时下降67%。正确用法二状态机复用。某实时风控规则引擎需对用户行为流做滑动窗口统计如“过去10分钟内登录失败次数5次则拦截”。若用map每次都要重建窗口状态用mapPartitions可复用状态机df.mapPartitions { iter val window new SlidingTimeWindow(10, TimeUnit.MINUTES) // 分区级复用 iter.map { event window.add(event.timestamp, event.type) val failCount window.count(login_fail) if (failCount 5) event.copy(risk_level HIGH) else event } }这里SlidingTimeWindow是自定义状态类避免了map中反复初始化的开销。致命陷阱闭包变量泄漏。mapPartitions的函数体在Driver端序列化发送到Executor。若函数引用了Driver端的大对象如一个1GB的机器学习模型整个对象会被复制到每个Executor。某次我们尝试在mapPartitions里加载XGBoost模型结果Driver内存直接爆掉。解决方案是改用Broadcast变量val modelBC spark.sparkContext.broadcast(model) // Driver端广播 df.mapPartitions { iter val model modelBC.value // Executor端获取 iter.map(event model.predict(event.features)) }3.4repartition与coalesce数据仓库里最常被滥用的“分区手术刀”这两个算子的名字极具迷惑性。“repartition”听起来像“重新分区”“coalesce”像“合并”但它们的物理行为截然不同且在数据仓库场景下选错等于给ETL链路埋雷。repartition(n)的本质是全量Shuffle。无论当前分区数是多少它都会触发Shuffle生成n个新分区。这在两种场景下是刚需解决数据倾斜当groupByKey后发现某个key的数据量是平均值的100倍必须用repartition(200)打散数据再聚合。适配下游并行度某报表任务需写入HDFS而HDFS块大小是128MB。若上游分区数据量不均有的200MB有的5MB会导致小文件问题。此时repartition(ceil(totalSize/128MB))能生成大小均衡的分区。coalesce(n)的本质是分区合并不触发Shuffle。它只能减少分区数且合并逻辑是“相邻分区合并”。这意味着当前分区数为100想减到50coalesce(50)可行当前分区数为100想增到200coalesce(200)无效分区数保持100当前分区数为100想减到30coalesce(30)会把前70个分区合并成30个但最后30个分区71-100保持不变导致数据分布不均。真实案例某日志分析任务上游是Kafka直连分区数Kafka topic分区数50。下游需要聚合统计我们误用coalesce(200)想增加并行度结果发现作业并行度仍是50且Shuffle Read量暴增——因为coalesce(200)无效Spark被迫用repartition兜底。正确做法是repartition(200)并配合spark.sql.adaptive.enabledtrue让自适应引擎动态调整。终极法则用explain看物理计划。任何对分区数的调整必须执行df.repartition(200).explain(true)在 Physical Plan 部分确认是否出现Exchange rangepartitioning表示Shuffle已触发rangepartitioning后的分区数是否等于目标值Shuffle Read Size是否在合理范围通常总数据量的1.5倍。我们曾为一个2TB的订单表做repartition(1000)explain显示Shuffle Read Size为3.2TB远超预期。根因是join key分布极不均匀导致range partitioning切分失衡。最终改用repartitionByRange配合自定义分区器把Shuffle Read Size压回2.1TB。4. 实操全流程从作业提交到线上稳定运行的完整链路4.1 开发阶段如何用explain和queryExecution揪出算子隐患在IDEA里写完Spark代码千万别急着提交。真正的调试从explain开始。我们团队强制要求每个新作业上线前必须提供三份explain输出df.explain(simple)看逻辑计划是否符合预期比如filter是否下推df.explain(extended)看物理计划里Shuffle类型、分区数、数据大小预估df.explain(cost)看Catalyst估算的成本对比不同写法的代价差异。案例排查一个诡异的性能倒退。某用户分群任务上周运行耗时18分钟本周突然涨到45分钟。explain(simple)显示逻辑计划一致但explain(extended)里发现关键差异// 上周 *(2) HashAggregate(keys[user_id#123], functions[count(1)]) - Exchange hashpartitioning(user_id#123, 200) // 本周 *(2) HashAggregate(keys[user_id#123], functions[count(1)]) - Exchange rangepartitioning(user_id#123, 200)hashpartitioningvsrangepartitioning——后者需要全局采样估算数据分布增加了采样耗时。根因是本周新增了一个orderBy操作触发了Catalyst的range partitioning策略。解决方案在agg前显式添加repartition(200)强制走hash partitioning。更深层的调试是查看queryExecution。在Spark Shell里执行val qe df.queryExecution println(qe.optimizedPlan.numberedTreeString) // 查看优化后逻辑计划 println(qe.executedPlan.numberedTreeString) // 查看实际执行计划executedPlan会暴露Catalyst的“妥协”——比如你写了broadcast(df_dim)但executedPlan里显示SortMergeJoin说明广播失败必须检查df_dim.count()是否超过阈值。开发阶段必做三件事在本地用--master local[4]模式跑小数据集用jstack抓取Executor线程栈确认无死锁用df.explain(formatted)生成可视化执行计划需Spark 3.3直观查看Stage依赖对所有join操作手动执行df_dim.count()确保小于spark.sql.autoBroadcastJoinThreshold。4.2 测试阶段用“压力探针”模拟生产数据分布数据仓库测试的最大陷阱是用均匀分布的测试数据验证逻辑。真实生产数据永远是长尾的——90%的用户只产生1%的行为10%的用户贡献90%的数据量。我们构建了一套“压力探针”机制Step 1生成倾斜数据。用spark.range(1000000)生成100万行但让其中1000行的key值相同val skewedKeys Seq.fill(1000)(SKEWED_KEY) val normalKeys (1 to 999000).map(_.toString) val allKeys (skewedKeys normalKeys).toSeq val testDF spark.sparkContext.parallelize(allKeys) .map(key (key, svalue_$key)) .toDF(key, value)这个数据集里SKEWED_KEY占比0.1%但数据量占比1000/10000000.1%——等等不对1000行占100万行的0.1%但groupByKey时这1000行全在一个分区而其他99.9万行分散在多个分区这才是真实的倾斜。Step 2注入生产级数据特征。真实日志有字段缺失、类型异常、编码错误。我们用withColumn注入testDF .withColumn(user_id, when(rand() 0.001, null).otherwise(col(user_id))) // 0.1%空值 .withColumn(event_time, from_unixtime((unix_timestamp() - 86400 * 30) rand() * 86400 * 30)) // 时间乱序 .withColumn(json_data, when(rand() 0.0001, invalid_json).otherwise(genJson())) // 0.01%脏数据Step 3运行时监控埋点。在关键算子后插入监控val monitoredDF testDF .mapPartitions { iter val counter TaskContext.get().taskMetrics().inputMetrics() iter.map { row counter.incRecordsRead(1) if (row.getString(0) SKEWED_KEY) { counter.incBytesRead(1024) // 模拟大记录 } row } }然后在Spark UI的Stages页签里查看Input Records和Input Bytes是否符合预期。若SKEWED_KEY的Input Bytes远高于其他key说明倾斜已复现。4.3 上线阶段灰度发布与熔断机制数据仓库作业上线不是“一键提交”而是分三步灰度Step 1小流量验证。用where dt20240101 and rand() 0.01随机采样1%数据验证逻辑正确性。重点检查count()结果是否与历史同口径一致允许±0.5%误差max(event_time)是否在合理时间范围内防止数据漂移explain的Shuffle Read Size是否与预估一致。Step 2全量预热。关闭spark.sql.adaptive.enabled用repartition(200)固定分区数跑全量数据。目的不是验证结果而是摸清资源水位记录Executor最大内存使用率目标75%记录Shuffle Write峰值目标集群总带宽的60%记录GC时间占比目标5%。Step 3自适应熔断。在作业启动时注入熔断逻辑val conf new SparkConf() .set(spark.sql.adaptive.enabled, true) .set(spark.sql.adaptive.coalescePartitions.enabled, true) .set(spark.sql.adaptive.skewJoin.enabled, true) // 自动检测倾斜Join .set(spark.sql.adaptive.localShuffleReader.enabled, true) // 熔断若Shuffle Read Size 10TB自动终止 val job spark.sql(SELECT /* REPARTITION(200) */ ...) job.foreachPartition { _ val metrics TaskContext.get().taskMetrics().shuffleReadMetrics() if (metrics.totalBytesRead 10L * 1024 * 1024 * 1024 * 1024) { throw new RuntimeException(Shuffle Read too large: metrics.totalBytesRead) } }这个熔断机制在某次数据源异常上游写入重复数据时触发避免了集群被拖垮。4.4 运维阶段基于Spark UI的“算子级”诊断手册Spark UI是数据仓库工程师的CT机但90%的人只看Jobs页签。真正的诊断在Stages和Storage页签Stages页签关键指标Duration单个Stage耗时。若某个Stage远长于其他如120s vs 平均8s点进去看TasksShuffle Read/Write单位是bytes。若Shuffle Read是Input Size的5倍以上说明Shuffle效率低下Records Read/Written对比Input Records若Shuffle Read Records远大于Input Records说明存在数据膨胀如笛卡尔积。Tasks页签深挖点击耗时最长的Task看Advanced里的JVM GC Time。若10s说明内存不足需调大spark.executor.memory看Input Size和Shuffle Read Size的比值。若比值0.5说明Shuffle读取了大量无关数据应检查filter是否下推看Locality Level。若大量Task显示NODE_LOCAL说明数据本地性好若多为PROCESS_LOCAL说明Executor内存不足数据被换出。Storage页签避坑Cached Partitions确认cache()的分区是否真的被缓存。若显示0/200说明缓存失败可能是内存不足或unpersist()被误调Storage Level确认是MEMORY_AND_DISK_SER而非MEMORY_ONLY后者在内存不足时直接丢弃导致重复计算。我们曾用Storage页签发现一个隐藏Bug某作业cache()后Storage显示198/200但下游作业报FileNotFoundException。根因是cache()时用了MEMORY_ONLY而Executor内存波动导致2个分区被驱逐但血缘系统未感知仍认为数据可用。强制改为MEMORY_AND_DISK_SER后问题消失。5. 常见问题与独家排查技巧实录5.1 “作业突然变慢”问题速查表现象可能原因排查命令解决方案Stage耗时突增300%Shuffle Read Size暴增explain(extended)看Shuffle Read检查filter是否失效或上游数据分布突变如新增倾斜keyExecutor频繁OOMmapPartitions里创建大对象jstack看线程栈找OutOfMemoryError改用Broadcast变量或在mapPartitions里用try...finally释放资源Shuffle Write速度骤降网络带宽打满top看Network进程iftop看网卡流量调小spark.shuffle.file.buffer默认32KB至16KB降低单次写入量Task长时间Pending数据本地性差Spark UIStages页签看Locality Level增加spark.locality.wait默认3s至10s容忍更长等待count()结果不准cache()后上游数据变更spark.catalog.clearCache()后重跑count()强制刷新缓存或改用checkpoint()保证一致性独家技巧用spark.sql.adaptive.enabledtrue自动修复。当发现Shuffle Read Size异常开启自适应后Spark会自动插入CoalescePartitions或AdaptiveSparkPlan优化。但要注意它只对Shuffle阶段生效对repartition无效。所以我们的标准动作是先repartition(200)再开启自适应让引擎在200个分区基础上做微调。5.2 “数据倾斜”问题的七种实战解法数据倾斜是数据仓库的头号杀手没有之一。我们总结了七种经生产验证的解法按优先级排序加盐Salting对倾斜key加随机前缀打散后聚合再去除前缀。适用于groupByKey、reduceByKey。某用户画像job对user_id加盐后倾斜key的处理时间从12分钟降至23秒。两阶段聚合第一阶段map时加随机前缀reduceByKey第二阶段去掉前缀再reduceByKey。比加盐更省内存但逻辑稍复杂。Broadcast Join对小维度表强制广播。必须配合spark.sql.autoBroadcastJoinThreshold调优。倾斜Key分离如前所述单独处理倾斜key再union结果。适用于join场景。增大Shuffle分区数spark.sql.shuffle.partitions400默认200。但要注意分区数过多会增加Task调度开销需平衡。使用repartitionByRange对有序key如时间戳用repartitionByRange替代repartition避免range partitioning采样。业务逻辑规避最彻底的方案。比如统计UV不用count(distinct user_id)改用approx_count_distinct(user_id)误差1%但性能提升10倍。避坑提醒所有加盐方案盐值必须是Int类型不能是String。因为String的hashCode()在不同JVM版本可能不同导致加盐后key分布不一致。我们统一用user_id.hashCode % 100生成盐值。5.3 “序列化异常”问题的根因分析与修复Task not serializable是Spark新手的噩梦但资深工程师知道它90%源于闭包变量引用。根因一Driver端对象未序列化。比如在map里引用了Spring Bean// 错误MyService是Spring管理的Bean无法序列化 val service applicationContext.getBean(classOf[MyService]) df.map(row service.process(row
