★ FEATURED ARTICLE 前言先说说我为什么想写这篇Spark介绍做了这么多年数据开发身边总有人问我“Spark到底是个什么东西”尤其是刚转行做数据工程的同学看到Spark和Hive、MapReduce、Flink这些概念堆在一起直接就晕了。我最早接触Spark的时候也一样翻文档看到RDD、DAG、Shuffle、Executor这些术语感觉每个字都认识连起来就是看不懂。后来在几个真实的离线数仓和实时链路项目里反复用才慢慢把这条线理顺。这篇文章不打算抄官方文档而是把我实际用下来的理解讲清楚Spark是什么、它解决了什么问题、核心运行原理长什么样、上手怎么写代码、以及那些文档里不会写清楚的调优和踩坑经验。不管你之前只写过SQL还是用过MapReduce但觉得太难用又或者是刚接手一个Spark作业想快速搞懂这篇文章应该都能给你一条清晰的上手路径。我会尽量把复杂的原理用大白话拆开讲也会附上可以直接跑的代码和参数配置你照着操作一遍就能有体感。1. 为什么需要Spark——它是为了解决什么问题而生的1.1 从MapReduce的痛点说起要说清楚Spark的价值得先说清楚它出现之前的那个时代。早年间做大数据批处理MapReduce基本是绕不开的选择Hadoop生态里大量离线作业都是靠MR跑起来的。MR的思想很朴素把任务拆成Map和Reduce两个阶段Map阶段并行处理原始数据把中间结果写到磁盘Reduce阶段再拉取这些中间结果做聚合。思路没有错但用起来是真痛苦。最让我受不了的是两点第一每个MR任务中间结果都要落盘一次复杂的ETL下来可能要串四五个MR任务每一轮都是读磁盘、写磁盘、再读磁盘。如果涉及迭代计算比如机器学习里反复训练模型这个代价简直离谱。第二写代码极其繁琐一个最简单的WordCount都要继承Mapper、Reducer类重写一堆方法再用Job对象把各个零件拼起来。业务逻辑没写几行胶水代码写了一大堆。Spark就是在这个背景下火起来的。它的核心理念是把中间结果尽量留在内存里借助DAG有向无环图调度引擎把多个计算阶段串起来能不走磁盘就不走磁盘同时提供一套比MR简洁得多的函数式编程接口。你用MapReduce需要几十行能跑通的事用Spark可能十行就结束了稍微有点代码基础的人半小时就能写出第一个作业。1.2 Spark到底解决了什么问题站在从业者的角度我习惯把Spark解决的问题归纳成三个方向。第一个是大规模数据的快速批处理。这个“大规模”没有绝对标准但从实践经验看单表几亿行、日志日增量几百GB到几个TB这个量级Spark跑起来很舒服。同样的数据量和同样的集群资源下实测Spark通常比相同逻辑的MR作业快数倍到十几倍尤其是两者都有Shuffle的复杂场景差距格外明显。第二个是统一计算引擎。Spark生态里包含Spark SQL、Spark Streaming、MLlib、GraphX这几个核心组件对应批处理、流处理、机器学习、图计算。这意味着同一套代码框架、同一个技术栈可以覆盖非常多的业务场景对团队来说学习成本和运维成本都更低。我见过不少团队只有批处理需求硬是引入了Spark后来发现SQL能力足够强直接把原来的Hive跑数任务迁过来了一批存量作业全被简化。第三个是易用性。早期版本的Spark支持Java、Scala、Python、R四种语言API后来R语言热度明显下降主流就是Scala和Python。你如果会SQL用Spark SQL基本零门槛你如果会Python用PySpark写业务也顺得很。对于数据分析团队来说用Spark做数据探索比直接写复杂SQL在某些场景下更灵活因为可以按行写逻辑、逐步验证这对调试和分析思路的帮助非常大。当然Spark也有自己的使用边界。它本质上还是面向批处理和准实时的框架默认的微批模式延迟是秒级到分钟级不可能替代毫秒级响应的在线实时系统。另外一个常见误区是拿Spark处理几百兆的小数据这属于杀鸡用牛刀集群调度开销比计算本身还大直接用Pandas或者其他单机工具反而更快。认清边界你才能选对工具。2. Spark核心架构与运行原理——从提交作业到算出结果中间发生了什么2.1 集群里的角色分工Driver、Master、Worker、Executor我第一次看Spark架构图的时候最迷惑的就是这些角色到底谁管谁。后来我自己画了一张图把集群想象成一个软件公司所有角色一目了然。Driver相当于项目负责人。它运行你提交的main函数负责解析代码、把一个作业拆成一系列任务、调度任务、汇总结果。它跑在提交作业的那台机器上Client模式或者在集群中单独申请一个容器Cluster模式。Master相当于公司的总监角色只管协调资源。真正干活的节点是Worker节点每个Worker节点上会启动若干个Executor进程。Executor是真正执行任务的员工。它是一个个JVM进程负责执行Driver派发过来的Task同时缓存你显式缓存的数据。每个Executor能同时跑多少个Task由它的CPU核心数决定。还有一个容易忽略的角色是Task这是Executor内部最小的执行单元。一个作业会被拆成多个Stage一个Stage又被拆成多个Task一个Task负责处理数据集的一个分区。我打个比方如果一个作业要统计100个文件的行数Driver会把数据划分成比如100个分区每个分区对应一个Task然后这些Task分发到各个Executor里并行执行。Executor越多、每个Executor的核心越多并行度就越高。你在spark-submit里配置executor数量和内存本质上就是在决定这个作业能同时雇多少员工干活。在实际工作中你不需要手动管理这些角色但必须理解它们的边界。比如Driver内存不够了日志会报Driver OOMExecutor内存不够报的就是Executor OOM。这两种问题的定位方向完全不同不理解角色分工排查效率会非常低。2.2 RDD、DAG与Stage划分——一个作业的生命周期Spark最底层的抽象是RDD弹性分布式数据集。理解RDD我建议抓住这三个关键词不可变、分区、惰性。不可变是说RDD一旦创建就不能修改你对它做任何转换操作比如Filter、Map都会生成一个新的RDD而不是修改旧的那个。这跟Scala的函数式风格一脉相承也保证了容错设计上的简便——每个RDD都通过血缘关系记录自己怎么来的某个分区丢了直接重新计算就行不用做数据备份。分区是并行计算的基础。一个RDD由多个分区组成每个分区会被一个Task处理。分区数通常由你读取的源数据决定比如读一个HDFS文件默认块大小128MB2GB的文件就有16个分区。你可以在后续操作里用repartition强行改变分区数这会影响并行度和最终文件数量。惰性求值是这个抽象最巧妙的地方。所有转换操作Transformation并不会真的立刻执行Spark只是记录了一个“待办清单”只有遇到Action操作比如collect、count、saveAsTextFile才会真正触发计算。这个过程有点像你用外卖App加了购物车真正下单付款的那一刻后厨才开始炒菜。这个设计让Spark的优化器有机会对计算过程做整体调优能合并的步骤合并能只算必要数据的只算必要数据。再说作业是怎么被拆分的。当Action触发计算后Driver会构建一张DAG有向无环图里面每个节点是一个RDD每条边是一个依赖关系。接下来DAG调度器会按依赖关系把图切分成多个Stage。切分的依据是宽依赖和窄依赖窄依赖是父RDD的每个分区只被子RDD的一个分区使用比如map、filter。这种依赖调度简单还支持流水线执行效率最高。宽依赖是父RDD的一个分区会被子RDD的多个分区使用典型的就是groupByKey、join、reduceByKey。宽依赖意味着数据要跨节点传输这就是经常听到的Shuffle。Stage的切分规则也很简单遇到宽依赖就在这里“结账”前面的所有窄依赖合成一个Stage后面再重新开一个Stage。一个经典的groupBy操作通常是两个Stage前一个Stage各分区自己计算key的数量后一个Stage把相同key的数据拉过来汇总。注意每个Stage之间会落盘窄依赖组成的Stage内部则通过管道式执行一气呵成。我在调优时特别注意一个细节Stage的数量和Shuffle的次数是决定作业性能的关键指标。同样一个关联逻辑如果两张表都做了合理预处理可能只需要一次Shuffle如果写的代码不讲究同样的结果可能会触发多次宽依赖多出来的那几次Shuffle就相当于多搬了好几次家代价非常大。3. 核心API选型与代码实操——RDD、DataFrame、Dataset到底怎么选3.1 三种抽象对比RDD、DataFrame、Dataset怎么选很多初学者第一个困扰就是刚接触Spark时看到三套API不知道学哪个。我用实际经验做一个明确的说法现在写业务绝大多数场景首选DataFrame不要一上来就写RDD。我把三者的关系整理成一张对比表一目了然对比维度RDDDataFrameDataset数据形式不结构化Java/Scala对象结构化带Schema结构化强类型对象适用语言Scala/Java/Python所有仅Scala/Java性能差依赖手动优化优自动通过Catalyst优化优自动优化Schema检查编译期不管运行期检查编译期检查易用性偏底层函数式API极强可直接写SQL偏强但需要类型编程Python支持支持支持不支持这份表看下来你会发现DataFrame在性能和易用性上全面占优尤其是它背后有Spark SQL的Catalyst优化器会自动做谓词下推、列剪枝这些优化而RDD阶段这些优化全靠你手工完成。数据量上了规模之后同样的逻辑用DataFrame写比用RDD写通常能快一倍以上这是优化器的功劳不是你的功劳。Dataset强类型在编译期就能发现字段名错误这才体现编译期检查的价值但在深度使用Python的情况下DataFrame是更现实的选择而且PySpark的DataFrame和Scala DataFrame在绝大多数能力上对齐学会SQL就等于会写Spark。那么RDD是不是就不用学了也不是。我建议你理解RDD的算子和机制因为它是Spark的地基理解了RDD你才能理解DAG是怎么生成的、宽窄依赖是什么、为什么某些算子会引发Shuffle。只是实际写业务逻辑的时候优先用DataFrame和SQL只有极少数场景比如需要直接从自定义数据源解析文本、没有现成Schema、或者要操作非结构化二进制数据我才会退回去写RDD。3.2 从零跑通一个WordCountWordCount是大数据领域的“HelloWorld”我直接给一份可以在本地环境跑通的PySpark代码每行都附上说明。from pyspark.sql import SparkSession # 1. 创建SparkSession这是新版Spark的统一入口 spark SparkSession.builder \ .appName(word_count_demo) \ .master(local[*]) \ .getOrCreate() # 2. 读取文本文件生成DataFrame df spark.read.text(file:///path/to/your/input.txt) # 3. 导入正则拆词把每行文本拆成单词 from pyspark.sql.functions import explode, split, lower, regexp_replace words_df df.select( explode(split(lower(df.value), r\s)).alias(word) ) # 4. 按单词分组统计数量 result words_df.groupBy(word).count().orderBy(count, ascendingFalse) # 5. 触发计算并输出这就是Action操作 result.show(20) # 6. 保存结果 result.write.mode(overwrite).csv(file:///path/to/your/output_dir) # 7. 收尾关闭 spark.stop()这段代码里每一步背后都有讲究。第1行创建SparkSession是整个程序的入口旧版里还要分别创建SparkContext和SQLContext现在统一成一个Session了。第2行用read.text读取原始文本生成一个单列DataFrame列名默认是value这一步并不会真的把文件全部加载进内存只是建立了读取的“待办清单”。第3行是整个作业的核心split把一行文本按空白字符切成单词数组explode把一个数组打散成多行一行的文本变成很多行单词。这两步合在一起就是RDD时代经典的flatMap操作。第4行做分组聚合第5行的show()才是真正触发整个Spark作业计算的开关。如果你环境里没有现成的PySpark可以先用pip install pyspark快速装一份。在本地跑这段代码不需要任何集群master(local[*])的意思是使用本机全部CPU核数模拟分布式环境调试代码非常方便。但注意仅仅是调试生产环境千万不要写死local模式要用spark-submit提交到集群。3.3 Spark SQL实战用写SQL的方式操作大数据我用了很久Spark之后越来越喜欢用Spark SQL处理日常任务因为逻辑更容易被别人理解。下面的示例演示实际项目中很常见的一类操作把订单表和用户表关联再做聚合。# 读取两张表数据 orders spark.read.json(hdfs://your-hdfs/orders.json) users spark.read.parquet(hdfs://your-hdfs/users.parquet) # 注册成临时表这步非常关键 orders.createOrReplaceTempView(orders) users.createOrReplaceTempView(users) # 直接用纯SQL写业务逻辑 result_sql spark.sql( SELECT u.country AS country, COUNT(DISTINCT o.user_id) AS user_cnt, SUM(o.amount) AS total_amount, AVG(o.amount) AS avg_amount FROM orders o LEFT JOIN users u ON o.user_id u.user_id WHERE o.order_date 2024-01-01 GROUP BY u.country ORDER BY total_amount DESC ) result_sql.show()这段代码的精髓在于你把表注册成视图之后剩下的所有处理都用纯SQL描述Spark的Catalyst优化器会帮你做关联顺序优化、条件过滤提前等一堆内部优化。你会发现文档里那些晦涩的“谓词下推”概念在这里变成了一件自然而然的事你在SQL里写了WHERE条件Spark会尽量在读取数据时就过滤而不是先全量读进来再扔。在实际项目中我通常把两种方式混着用能一行SQL讲清楚的逻辑直接用SQL需要写循环、按行自定义函数处理的逻辑用DataFrame API。这两种方式在同一个SparkSession里可以自由切换临时表名就是桥梁。另外提醒一句注册视图的时候命名要规范一个复杂作业里会注册好几张临时视图名字乱七八糟到后面自己都看不懂。3.4 常用算子与性能细节这几个坑我帮你踩过了这部分把几个高频算子拿出来对比每一个都是实际作业里反复用到、而且换一种写法性能天差地别的场景。groupByKey和reduceByKey永远选reduceByKey。两者都能做分组聚合但groupByKey会直接把每个key对应的所有value都原样shuffle到同一个节点如果有100万个value落到同一个key上这100万个数据都要在网络里传输一遍内存也扛不住。reduceByKey会在shuffle之前先在每个分区内部做一次预聚合比如求和的话每个分区先把这个key的那份和算好了再跨节点合并整个网络传输量少了一个数量级。我处理日志统计时同样一个逻辑用reduceByKey改写后作业运行时间从半小时降到十分钟。如果当前任务用到了groupByKey几乎可以肯定能优化。coalesce和repartition两者都是调整分区数但代价完全不同。repartition是真正做一次shuffle把数据打散到指定的分区数coalesce是合并分区尽量避免shuffle只能用于减少分区数。实际场景里跑完大作业要把结果写回HDFS时如果分区数太多写出的文件会又碎又多所以我经常用coalesce把分区合并到合理数量再落盘。反过来当你发现数据量不小但分区太少、并行度不够时只能repartition来增加分区。一句话记住减少分区优先coalesce增加分区只能repartition。手动触发缓存要冷静。缓存cache/persist的作用是让某个数据在之后的多个Action里复用避免重复计算。但用了它不代表万事大吉因为缓存占的是Executor内存缓存太多数据会把其他计算的内存挤没反而出现OOM。我有个朴素的判断原则只在同一个DataFrame会被反复使用、且重新计算的代价明显偏大时才显式缓存比如一个耗费十几分钟的宽表作业后续要基于它做多类统计。一次性使用场景缓存就是白占内存。4. 性能调优实战——从“能跑”到“快跑”的关键调整4.1 资源参数怎么定一套可以复用的计算逻辑这一步是最不直观的因为网上文档只告诉你有哪些参数没告诉你该怎么给数值。我分享一个自己常用的配置推导逻辑以Yarn模式、集群单节点32核64GB内存为例。先说Executor数量。Yarn能分配给Spark的总资源是有限度的每台Worker节点上驻留的Executor数不要过多因为Executor是JVM进程需要内存开销。我的经验值是单节点最多驻留3到4个Executor每个Executor分配4~5个Core。比如你的集群有10个可用节点每个节点贡献4个Core总数就是40个CoreExecutor数量按每个Executor4个Core计算大约可以开10个Executor。再算Executor内存。一个Executor默认分配的内存包括执行内存和存储内存Spark内存模型里这两者又按比例分治。通常我给每个Executor配8GB左右堆内执行内存默认占60%左右这个比例在多数作业里够用。然后是三个最常调的参数参数作用我的常用值spark.executor.instancesExecutor个数由总Core数除以单Executor Core数得到spark.executor.cores单个Executor的核心数4或5spark.executor.memory单个Executor堆内内存8G或16G取决于单节点总内存spark.sql.shuffle.partitions跑SQL时Shuffle的分区数默认200大作业调到500~1000最后一个参数是最容易出问题的。Spark SQL执行Shuffle默认按200个分区来如果你的数据量特别大200个分区意味着每个分区的Task要处理的数据非常多并行度严重不足。反过来如果数据量只有几百MB200个分区又太碎调度开销比计算大。我的经验是让每个分区数据量在100MB到1GB之间比如你的数据约2GB的临时结果要Shuffle分区数设置在64左右比较合适。这个估算规则在很多作业里都通用。4.2 数据倾斜怎么处理遇到慢任务就不再玄学数据倾斜是Spark作业调优里最常见的硬仗特征是运行日志里某个Stage的任务绝大多数几秒就完成了但有那么一个或几个任务跑了十几分钟甚至半小时整个作业被这个长尾任务拖死。本质是数据分布不均某个Key的值特别多或者某个分区的数据量特别巨大。我把日常用的三种处理手段按优先级写在这里。第一种如果倾斜是由两个大表Join引起的优先考察能不能用广播小表。所谓广播就是把一个比较小的表完整发送到每个Executor的内存里这样Join的时候就不需要Shuffle那个大表的全量数据。Spark广播的阈值默认是10MB你可以通过spark.sql.autoBroadcastJoinThreshold调大比如把判断依据改成50MB甚至100MB。实际操作中如果一个表经过过滤后只剩几十MB广播它比走正常Shuffle要快非常非常多。第二种如果不是小表倾斜只发生在少数Key上就用加盐拆分。办法是把一个倾斜的大Key加上随机前缀打散成多个小Key另一张表做无条件复制按前缀拼接匹配。举个例子假设user_id1的用户有80%的订单数据你可以在Map端把这个大Key加0到9十个随机后缀然后在另一张表里给这个用户复制成10份、分别拼接后缀让它们分别落在不同分区上Join最后再按原始Key聚合回来。这套逻辑虽然要多写几步但几乎能根治所有单Key倾斜问题。第三种如果倾斜的原因是某个分区在Shuffle后数据量超预期而业务上对聚合结果没有极致的精确要求可以考虑两阶段聚合先按Key加随机后缀做一次局部聚合再去掉后缀做一次全局聚合。这个方法对groupBy类型的倾斜非常高效。我踩过一个坑是最开始遇到倾斜动不动就想加随机后缀结果发现很多情况下只是Shuffle分区数设置太少导致的伪倾斜把spark.sql.shuffle.partitions从200调到800就解决了。所以我现在的排查顺序是先看UI确认是否真的存在长尾任务再查分区数是否合理再考虑广播最后才是加盐改造。先简单后复杂别一上来就大动干戈。4.3 内存与缓存策略堆内、堆外要分清Spark内存相关的概念容易被搞混这里必须说清楚Executor内存不只是一块JVM堆内空间而是分成堆内和堆外两部分。堆内是你通过spark.executor.memory设置的部分堆外内存通过spark.memory.offHeap.size设置默认不开需要显式开启。在默认配置里Spark会用spark.memory.fraction默认0.6把堆内内存切成两半执行内存和存储内存。执行内存用于Shuffle时的聚合排序存储内存用来缓存数据。这两者还可以互相占用当只有计算没有缓存时执行内存可以尽量多吃空间。所以我做生产作业的时候除非明确要反复复用数据集否则不会开持久化缓存目的就是让执行内存有足够地盘可用。缓存级别也有学问。cache()等价于persist(StorageLevel.MEMORY_ONLY)数据只放内存放不下就重新计算。对很多需要迭代计算的场景我建议用MEMORY_AND_DISK内存放不下会溢写到磁盘不会直接丢失重算。要注意的是persist之后数据还在内存里不会自动清理直到任务结束或你手动unpersist()。我用的时候通常在每个Action跑完后就unpersist避免长时间占据存储内存。内存分配没有放之四海皆准的公式但有一个观测工具你一定得掌握在Spark UI的Storage页签里看每个RDD/DataFrame的缓存占比在Executors页签看每个Executor的Used Memory和GC时间。那个页面就在那里只是太多人看了两眼就关掉实际上它是性能问题的第一手情报来源。5. 常见问题与排查经验——那些年我踩过的Spark坑5.1 Executor OOM先分清是哪种OOM再动手Spark作业挂在OOM上的频率估计做过的人都深有体会。但OOM和OOM不一样我建议第一步永远是看日志到底报的是哪边。如果是Executor OOM多数原因是单个Executor内存不够或者Shuffle数据量太大撑爆了执行内存报错信息里通常能定位到是哪个Stage和哪个Task。一般的处理顺序是先看这个Stage的输入数据量有多大如果真的很大优先增加分区数降低单Task压力这个改动最小也最有效如果数据量正常但Shuffle很大就要检查是不是算子写成了groupByKey而不是reduceByKey有没有在Shuffle前做预聚合最后才考虑调大spark.executor.memory因为加内存总会拖慢整个集群的资源周转。如果是Driver OOM往往发生在collect()把大数据全量拉回Driver的时候。这个错误在开发调试时特别常见因为大家喜欢collect来看结果但数据量大到一定程度Driver单机的内存根本装不下。解决办法是不要collect全量用take(100)看一下抽样结果或者写入磁盘再查验。我自己定位OOM问题有个笨办法在Spark UI里点进失败的那个Stage看输入数据量、Shuffle写量、Shuffle读量这三张表。哪个数据量异常大问题就基本锁定在哪一步。这套方法很老但每一次都能帮我快速逼近问题核心。5.2 序列化错误与依赖缺失Spark作业在集群运行时Driver、Executor是不同的JVM进程数据要在网络上传输所有对象必须能序列化。如果不做任何设置Spark默认用Java序列化器在代码里用了自定义类却没有实现Serializable接口就会在某个奇怪的阶段抛出Task not serializable异常。这个错误在Java和Scala项目里特别常见Python反而因为序列化机制不同而很少遇到。解决思路有两个层面一是让类实现Serializable接口通常就能解决二是把不需要跨进程传递的变量从类里抽出去尤其是那些持有数据库连接、文件句柄的成员变量它们天然不可序列化。我早年踩过一个坑在Spark代码里把一个服务类实例声明成主类成员变量结果每次提交作业都报序列化错误后来改成在Map函数内部创建才通过。另一个常见问题是Executor上缺少第三方依赖。比如你本地用Java写了一个封装依赖了某个工具库本地能跑提交到集群就报ClassNotFoundException。这个问题的解决方式很标准用spark-submit --jars带上外部jar包注意Spark的jar包机制和Java ClassLoader绑定很深包多了要谨慎版本冲突也能让人崩溃。最好在项目一开始就用构建工具统一管理依赖并打成一个fat jar从源头避免“本地能跑、集群不能跑”的尴尬。5.3 输出文件小文件爆炸Spark写HDFS或者Hive表时如果最后一步Spark作业的分区数特别多哪怕每个分区的数据只有几KB也会写出一堆小文件。小文件对于HDFS是灾难因为NameNode内存承载有限文件多到一定程度会拖垮整个集群。我在一个配置不合理的存量作业里跑一次日批任务写了两千多个大小是几KB的文件下游Hive查询慢得可怕先去合并小文件就花了半天。根治办法是在写操作前用repartition(期望分区数)或者coalesce(期望分区数)把分区数压下来。期望分区数怎么定先把目标数据总量算出来比如预计输出5GB数据每个输出文件按256MB算目标分区数就是20个左右。如果本来数据量就不大直接coalesce(1)写成一个文件也不是不行但注意压缩运算会变成单TaskRedis资源白白浪费所以这个做法只适合超小数据量。还有一个衍生问题是下游Hive表读不到数据因为Spark写了文件却没有更新Hive元数据。这不是小文件问题是元数据同步问题但和文件写出的方式强相关。解决办法是用insertInto或者通过DataFrame写入Hive表目录后再做一次MSCK REPAIR TABLE或者干脆用Spark SQL的INSERT OVERWRITE TABLE语法而不是文件API写文件。5.4 一个可直接复用的Spark提交脚本长什么样在实际项目里我很少直接敲spark-submit那一长串命令都是写一个脚本固定下来。下面这个模板可以套用到大多数离线作业里。#!/bin/bash APP_NAMEetl_job_name MAIN_CLASScom.example.EtlJob JAR_PATH/data/apps/spark-jobs/etl-job.jar INPUT_PATH/user/hive/warehouse/orders OUTPUT_PATH/user/hive/warehouse/orders_daily QUEUE_NAMEdefault spark-submit \ --master yarn \ --deploy-mode cluster \ --name ${APP_NAME} \ --class ${MAIN_CLASS} \ --queue ${QUEUE_NAME} \ --executor-memory 8G \ --driver-memory 4G \ --executor-cores 4 \ --num-executors 20 \ --conf spark.sql.shuffle.partitions400 \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.sql.adaptive.coalescePartitions.enabledtrue \ --conf spark.dynamicAllocation.enabledfalse \ --jars /data/libs/commons-lang3.jar \ ${JAR_PATH} \ ${INPUT_PATH} ${OUTPUT_PATH}有几个细节值得单独说。--deploy-mode cluster表示Driver在集群里运行而不是在你提交作业的客户端机器上生产环境推荐这个模式否则你的电脑一断网关掉浏览器作业就悬了。spark.sql.adaptive.enabledtrue是Spark 3.0开始引入的自适应查询优化能力能根据运行时数据情况动态调整分区数强烈建议开启。spark.dynamicAllocation.enabledfalse是我故意关掉的动态资源分配因为离线作业资源事先规划好更稳定动态分配在部分云环境里反而会引发资源争抢。5.5 常见问题速查表现象可能原因排查方向某个Stage一直卡着不走数据倾斜长尾TaskSpark UI看该Stage各Task耗时分布全作业很慢但无单个长尾分区数太少或资源不足检查executor数量、shuffle分区数报Executors OOMShuffle数据量过大或内存配置不足先加分区再检查预聚合算子报Driver OOMcollect全量数据改成take抽样或写文件Task not serializable自定义类未序列化、闭包捕获了不可序列化对象实现Serializable接口、移除闭包内的连接对象ClassNotFoundExceptionExecutor上缺依赖用--jars或打fat jar写文件产生海量小文件最终分区数太多写之前repartition/coalesce这张表不是什么高深理论都是我工作中一行日志一个UI界面排查出来的实际经验建议把它收藏起来遇到问题先对号入座。结尾我的几条实在建议如果你正准备上手Spark我的建议是先别急着啃架构原理拿PySpark在本地把WordCount和几个聚合分析跑通对“转换是惰性的”“Action才真正执行”产生体感之后再回头看那些原理你会突然发现以前看不进去的文档全都能看懂了。正式写生产作业时我自己的习惯是能写SQL就不写代码能用DataFrame就不写RDD把精力花在理解数据模型上而不是跟底层算子较劲。毕竟Spark的价值是帮你处理数据不是帮你炫技。最后分享一个小技巧每次作业跑完后养成看Spark UI的习惯哪怕只是花一分钟扫一眼每个Stage的耗时柱状图。哪个Stage慢就把鼠标移上去看看Shuffle读和写的量数据量异常的Stage通常就是优化的突破口。坚持个把月你对Spark性能的直觉会明显提升很多问题也能在提交前就提前避开。
阅读完成 · 觉得有帮助？
