Spark on Yarn实战避坑指南：内存模型、资源分配与外部数据源适配

做数据开发的人应该都有类似经历同样的代码在本地跑得飞快一提交到集群就各种怪问题日志翻半天也找不到头绪。这个 Scala 和 Spark 大数据分析系列写到第七篇我不打算再从头讲语法了而是把这一段时间在项目里反复踩过的环境搭建、内存配置、Yarn 资源分配、外部数据源集成的坑一次性梳理清楚。无论你是刚刚装好 Scala 还没跑通第一个 Spark 任务的初学者还是已经在 Spark on Yarn 上被 CPU 核数和 Executor 内存折腾过的老手这篇的内容你应该都用得上。这个系列一直在强调一个观点Scala 和 Spark 天生就是一对Scala 的函数式编程风格和 Spark 的分布式计算模型配合起来写出来的分析代码既简洁又不容易出低级错误。这一篇的内容主要围绕几个热点问题展开Scala 安装时 Coursier 下载慢怎么处理、Spark 集群怎么搭才对、Spark 内存模型怎么理解、Spark on Yarn 作业为什么每个 Executor 只拿到一个 vCore、达梦数据库这类外部数据源怎么和 Spark 适配最后再聊一聊 Spark 面试里那些高频知识点。内容比较杂但都是实际项目里真正会碰到的问题。1. 为什么这个系列坚持用 Scala 写 Spark而不是 Python这个系列七篇写下来后台一直有读者问现在 PySpark 也成熟了为什么还要用 Scala这个问题我每次都要解释一遍今天在这里统一说说我的看法。1.1 Scala 和 Spark 的底层关系Spark 本身是用 Scala 写的它的核心 RDD、DataFrame、Dataset 的底层实现都是 Scala。这意味着你用 Scala 写 Spark等于直接和框架底层打交道很多 Python 环境下被封装掉的问题在 Scala 下能看得更清楚。比如 RDD 的依赖关系、Shuffle 过程的实现细节、Catalyst 优化器的一些行为用 Scala 调试时可以直接看源码PySpark 里很多时候只能看到一个黑盒子。另外还有一个实际好处Scala 是强类型语言配合 IDE比如 IntelliJ IDEA做自动补全和类型检查很多错误在编译阶段就暴露了不用等到集群跑起来才发现。对于复杂的 ETL 逻辑这个优势非常明显。我做过的几个大型数据处理项目代码量上万行的 Scala 作业出运行时类型错误的概率比动态语言低很多。1.2 项目选型时 Scala 和 Python 怎么权衡话说回来我也不是无脑推荐 Scala。如果是快速验证想法、做一些探索性的数据分析PySpark 确实更灵活。但如果是生产环境的稳定作业、要长期维护的数据平台、性能要求高的核心链路Scala 仍然是更稳妥的选择。我自己在项目里的选型标准是这样数据量在百 GB 以下、以临时分析为主用 PySpark 比较合适开发快。数据链路要长期跑、要接调度系统、要做单元测试和代码审查优先 Scala。团队里如果都是 Python 背景硬上 Scala 学习成本很高这种情况可以先 PySpark但一定要做好性能压测。用 Scala 写 Spark 还有一个容易忽略的好处可以无缝使用 Spark 的 Dataset API。Dataset 在编译期就能知道字段类型比 DataFrame 更安全做复杂业务时不容易因为字段名写错而翻车。这点后面写案例的时候会具体演示。2. 开发环境搭建这次把坑一次说完新项目开工第一步就是搭环境。这个环节看起来简单实际上一堆细节没处理好后面每一步都会卡壳。这一节我把 Scala 安装和 Spark 集群搭建的常见坑都过一遍。2.1 Scala 安装Coursier 卡在下载那里怎么办最近几年 Scala 官方推荐的安装方式是使用 Coursier 这个工具一条命令就能装好 Scala 和相关的编译工具。但在实际使用中很多人卡在第一步Coursier 自己要从 Maven 中央仓库拉东西网络一不稳定就卡住进度条纹丝不动。我遇到过的典型场景是执行完cs setup之后界面一直停在下载某个依赖等半小时都不动。这种问题的根源是 Coursier 默认使用了 Maven Central而某些网络环境下访问这个仓库很慢。解决办法很简单给 Coursier 配置一个国内镜像仓库。以阿里云 Maven 镜像为例在终端设置环境变量export COURSIER_REPOSITORIEShttps://maven.aliyun.com/repository/public然后重新执行cs setup就可以了。如果还有某些依赖拉不下来还可以把镜像地址追加成多个用逗号分隔。安装完成后记得验证scala -version能看到版本号就说明环境基本 OK。2.2 Spark 集群搭建前要想清楚的三件事Spark 本身不提供集群管理能力它需要一个外部资源管理器来分配资源。最常见的三种模式是 Standalone、Yarn 和 Kubernetes。很多人一上来就照着网上的教程搭 Standalone 集群结果后面才发现根本不适合自己的场景。我的建议是先想清楚这三件事再动手集群上是否还跑别的计算框架比如 Hive、MapReduce如果跑就用 Yarn方便统一管理资源。是否要用到动态资源分配如果用Yarn 模式支持得最成熟。是否要上生产环境如果只是本地学习直接用 Spark 自带的 Standalone 就够了部署简单。以 Yarn 模式为例搭建时的核心配置在spark-defaults.conf里。最常见的几个参数是spark.masteryarn spark.yarn.archivehdfs:///spark-libs/spark-archive.zipspark.yarn.archive这个参数容易被忽略。它是把 Spark 运行需要的 jar 包打成压缩包放到 HDFS 上让 Yarn 上的每个 NodeManager 去拉取。如果不配置每次提交作业都要把 jar 包随任务分发一次作业启动会非常慢任务一多整个集群都会被搞得很卡。2.3 Spark 内存模型配置错了跑再慢都不知道为什么在 Spark 1.6 之前执行内存和存储内存是分开固定配比的空间不能互相借用经常出现一边不够用一边在浪费。1.6 之后引入了统一内存管理一张图就能说明白Executor 内存被分成三块Reserved Memory、User Memory、Spark Memory而 Spark Memory 内部又分为 Execution Memory执行内存和 Storage Memory存储内存两边可以互相借用。实际项目里最容易踩坑的参数有三个spark.memory.fraction默认 0.6表示 Spark Memory 占整个 Executor JVM 堆的比例。调大这个值会让执行可用内存变多但留给用户代码的 User Memory 就少了。spark.memory.storageFraction默认 0.5表示 Storage Memory 在 Spark Memory 中的初始比例。RDD 缓存多的任务可以调大Shuffle 密集的任务可以调小。spark.executor.memoryOverhead默认值是max(executorMemory * 0.1, 384m)这部分是 Executor 在 JVM 堆之外额外申请的容器内存用来跑 JVM 线程、NIO 缓冲、本地临时文件等等。很多人只配了spark.executor.memory却忽略了这个 overhead导致容器内存超出 Yarn 限制任务被直接 kill 掉。参数默认值作用调优建议spark.executor.memory1GExecutor JVM 堆内存结合数据量调太大容易触发 Yarn 单容器上限spark.memory.fraction0.6执行与存储共用内存占比代码逻辑简单时可调大到 0.7spark.memory.storageFraction0.5存储内存初始占比缓存多调大Shuffle 多调小spark.executor.memoryOverheadmax(0.1*executorMemory, 384m)JVM 堆外额外内存会加入 Yarn 容器总内存必须考虑再补充一句Yarn 模式下每个 Executor 向 Yarn 申请的总内存等于spark.executor.memory加上spark.executor.memoryOverhead。如果这两个值加起来超过了yarn.scheduler.maximum-allocation-mb作业会直接报错启动失败。这个细节很容易被忽视但排查起来特别费时间。3. 一个可以直接复用的 Spark SQL 分析案例环境搭好、概念理清接下来用一个完整的案例把上面的知识串起来。这个案例不是教科书里那种刻意简化的小 demo而是我在项目里实际处理过的一类需求模拟了一个电商订单明细分析场景。3.1 场景与数据准备假设有一张订单明细表存在 Hive 里每天新增几百万条订单记录表里有订单号、用户 ID、商品类目、商品单价、下单数量、订单金额、订单状态、支付时间、省份城市等字段。要计算的指标包括每日订单总量、总销售额、Top 销售额商品类目、用户复购率等。先写一段 Scala 代码创建 SparkSessionimport org.apache.spark.sql.SparkSession val spark SparkSession.builder() .appName(OrderAnalysis) .enableHiveSupport() .config(spark.sql.shuffle.partitions, 200) .getOrCreate()spark.sql.shuffle.partitions这个参数是 Spark SQL 执行 Shuffle 时默认的分区数默认值是 200。很多初学者跑完代码只会看结果根本不关注 Shuffle 分区数结果就是数据量明明不大却开了几百个 Task资源浪费严重。后面调优章节会展开讲。3.2 数据清洗与质量检查拿到原始数据第一步不是算指标而是做质量检查。我用 Scala 写了一个简单的检查逻辑import org.apache.spark.sql.functions._ val orders spark.table(default.orders) // 基本数量检查 println(s总记录数: ${orders.count()}) // 关键字段空值检查 orders.select( count(*).as(total), sum(when(col(order_id).isNull, 1).otherwise(0)).as(null_order_id), sum(when(col(amount).isNull, 1).otherwise(0)).as(null_amount), sum(when(col(pay_time).isNull, 1).otherwise(0)).as(null_pay_time) ).show()清洗的核心思路是先搞清楚有哪些脏数据再决定是过滤掉、修复还是保留。比如支付时间为空的订单可能是未支付也可能是数据采集延迟如果直接过滤会导致销售金额偏低需要和业务方确认规则。订单金额为负数的情况大概率是退款单要不要计入指标也要先对齐口径。3.3 业务指标计算从 SQL 到 DataFrame API清洗完成后开始算指标。我通常先用 SQL 写一遍确认业务口径没问题再用 DataFrame API 重构因为 DataFrame API 更容易做单元测试和复用。比如计算每日销售总额和订单量val dailyStats orders .filter(col(status) paid) .groupBy(to_date(col(pay_time)).as(day)) .agg( sum(amount).as(total_amount), countDistinct(order_id).as(order_cnt) ) .orderBy(day) dailyStats.show(10)这里用了countDistinct在大数据量下代价很高因为要去重。很多场景其实可以用sum(1)配合 groupBy 替代或者提前在数据仓库层做好去重。如果真的要精确去重就要接受它带来的 Shuffle 成本。3.4 用 Dataset 的坑Encoder 与序列化DataFrame 用起来方便但它是弱类型的字段类型只有在运行期才能发现错误。生产作业我更喜欢用 Dataset。定义一个 case classcase class Order( orderId: String, userId: String, category: String, amount: Double, payTime: String, status: String ) import spark.implicits._ val orderDs spark.table(default.orders).as[Order]Dataset 能在编译期检查字段写复杂逻辑时不容易因为字段名拼错而出问题。但这里有个经典的坑case class 只能定义在对象外面不能定义在方法内部或者是测试类内部否则 Spark 会报Unable to find encoder for type stored in a Dataset。这个错误几乎每个用 Dataset 的人都遇到过本质是编码器创建时找不到类的元数据。解决办法是所有用于 Dataset 的 case class 都放到独立的 Scala 文件中放在包的最外层。虽然看起来多写了一个文件但长期维护下来省心很多。4. Spark on Yarn 提交作业CPU 核数为什么提不上去这一节专门说一个被问烂了的问题同一个 Spark 作业在 Standalone 模式下能用到多个 CPU一提交到 Yarn 上每个 Executor 只分配到一个 vCore怎么调都没用。这个问题的热度一直很高说明踩坑的人特别多。4.1 现象与排查思路先描述一下现象。用spark-submit提交作业指定了--executor-cores 4日志里 Executor 也显示出来了但看 Yarn ResourceManager 界面每个 Container 的 vCore 还是 1。很多人在这里就开始怀疑是不是 Yarn 配置有错其实问题往往出在更隐蔽的地方。排查思路按顺序来确认提交命令里的参数有没有真正生效。可以用spark-submit --verbose查看最终传给集群的参数。确认 Yarn 调度器的配置。看yarn-site.xml里的yarn.nodemanager.resource.cpu-vcores和yarn.scheduler.maximum-allocation-vcores。确认 Spark 的spark.executor.cores有没有被其他配置覆盖。最常见的情况是yarn.scheduler.maximum-allocation-vcores配得非常小比如 2那即使应用请求 4 个 vCoreYarn 也会把它掐到 2 以下。这个参数是 Yarn 层面单容器可以申请的最大 vCore 数是硬限制。4.2 为什么默认会用 1 个 vCore这里还要解释一个更深层的原因。Spark 在很多发行版里的默认配置中spark.executor.cores的值是 1。也就是说如果用户在提交命令里没有显式指定--executor-cores或者只在代码里配置了 SparkSession 而没传参数Executor 启动时就只会申请 1 个 vCore。另外一个相关参数是spark.task.cpus默认也是 1。它表示每个 Task 需要占用的 CPU 核数。如果机器 CPU 资源比较充裕、作业又是纯 IO 密集型可以保持 1。但如果 Task 本身涉及大量 CPU 计算适当调大spark.task.cpus可以减少调度开销代价是并行的 Task 数量变少。我把常见的资源参数整理成一张表建议截图保存参数默认值含义常见坑spark.executor.cores1每个 Executor 占用的 CPU 核数不显式设置时 Executor 只拿 1 核spark.task.cpus1每个 Task 分配的核数调大后 Task 并行度降低spark.executor.memory1GExecutor 堆内存忘记加 memoryOverhead 导致容器超限spark.executor.instances无静态 Executor 数量与动态分配冲突时可能被覆盖yarn.scheduler.maximum-allocation-vcores默认较大Yarn 单容器最大 vCore配置过小时应用资源被截断yarn.scheduler.maximum-allocation-mb默认较大Yarn 单容器最大内存同时影响 Executor 和 Driver 的内存申请4.3 一个合理的内存核数配比到底怎么算关于 Executor 内存和核数的配比没有一个万能公式但有一个基本的安全准则单个 Executor 的核数不要太多。很多人觉得核数越多跑得越快其实未必。每个 Executor 是一个 JVM 实例核数多意味着 JVM 内的并发线程多GC 压力会变大而且 Executor 崩溃时影响面也更大。我一般这样配单个 Executor 的核数在 2 到 5 之间内存按每核 4G 到 8G 来估算。举个例子如果一台 NodeManager 机器有 16 核、64G 内存Yarn 能分配给 Spark 的资源按 75% 算就是 12 核、48G。可以这样分spark-submit \ --master yarn \ --deploy-mode cluster \ --num-executors 3 \ --executor-cores 4 \ --executor-memory 15G \ --driver-memory 4G \ --class com.example.OrderAnalysis \ order-etl.jar这里每台机器 4 核 15G 一个 Executor3 台机器一共 3 个 Executor。15G 不要顶满 16G留一点给 NodeManager 本身和系统开销。这只是个示例真实场景还要结合数据量、Shuffle 量来判断。4.4 动态资源分配到底要不要开Spark 默认的 Executor 数量是静态的作业提交时就一次性申请完毕。如果任务有波峰波谷静态分配会浪费资源。开启动态分配后Spark 可以根据任务积压情况动态申请和释放 Executor。配置方式spark.dynamicAllocation.enabledtrue spark.dynamicAllocation.initialExecutors2 spark.dynamicAllocation.minExecutors2 spark.dynamicAllocation.maxExecutors20 spark.shuffle.service.enabledtrue注意在 Yarn 模式下开启动态分配必须同时开启spark.shuffle.service.enabled。原因是 Executor 被释放后它产生的 Shuffle 中间文件还需要被后续 Task 读取如果没有外部 shuffle service 托底就要把 Executor 保留到 Shuffle 结束动态分配的意义就没了。但动态分配也不是银弹。如果作业本身是稳定的定时任务数据量波动不大静态分配反而更可控。我的经验是C 端业务、流量有明显波峰的场景适合动态分配B 端报表、每日定时跑批的场景静态分配配合合理的重试策略更省心。5. 排错实录日志、OOM 和外部数据源适配这一节把近期群里问得最多的几个报错和处理方法集中过一遍。每一个都是我实际排查过的问题按现场现象、排查思路、最终方案三个角度来写。5.1 log4j 警告怎么才能彻底消除作业一启动控制台或者日志里就会出现一行Using Sparks default log4j profile: org/apache/spark/log4j-defaults.properties这行日志不是错误但看着很扎眼。它的意思是 Spark 没有找到用户自定义的 log4j 配置所以退回到了默认配置。如果要自定义日志级别可以在$SPARK_HOME/conf目录下新建log4j2.propertiesSpark 3.x 使用 log4j2然后在spark-submit时用--files把配置带上去。还有一个隐蔽的问题如果集群环境里 classpath 出现了多个 log4j 实现Spark 会发出冲突警告导致日志打印混乱。解决方式是用--driver-class-path和--executor-extra-class-path显式指定依赖顺序或者在打包时排除掉不需要的 log4j 版本。spark-submit \ --files /path/to/log4j2.properties \ --conf spark.driver.extraJavaOptions-Dlog4j2.configurationFilelog4j2.properties \ --conf spark.executor.extraJavaOptions-Dlog4j2.configurationFilelog4j2.properties \ ...这样基本可以保证日志按你的配置输出不会出现被默认配置覆盖的问题。5.2 Executor 为什么会 OOM先别急着加内存OOM 是 Spark 作业里最常见的报错之一但很多人一看到 OOM第一反应就是加内存。这个思路有时候有效更多时候是舍本逐末。Executor OOM 通常分两种堆内内存不足报错里能看到java.lang.OutOfMemoryError: Java heap space。堆外内存不足报错里能看到Container killed by YARN for exceeding memory limits。第一种情况先检查是不是真正需要这么多内存。我之前遇到过一个大作业频繁堆内存溢出最后发现是因为读取的数据源文件格式是 text然后使用map算子做了大量字符串拼接产生了很多中间对象。改成用 DataFrame 和col()表达式后内存占用立刻降了一半。第二种情况大概率是spark.executor.memoryOverhead配置太小调整这个参数加堆外内存比盲目加大spark.executor.memory更有效。再补充一种比较隐蔽的spark.sql.autoBroadcastJoinThreshold默认是 10M如果小表刚好超过这个阈值Spark 就不再走 Broadcast Join而是退化成 SortMergeJoin。SortMergeJoin 会带来大量的 Shuffle 和排序内存开销成倍增加。遇到 join 慢和 OOM除了调内存还应该看看执行计划里两个表是 Broadcast 还是 Shuffle。5.3 达梦数据库与 Spark 适配的落地姿势近期有一个高频问题项目里用了达梦数据库想用 Spark 做批量分析应该怎么对接。这个需求在信创环境下尤其常见。其实达梦数据库兼容 PostgreSQL 协议的一部分但它提供的 JDBC 驱动是独立的不能直接用 PostgreSQL 驱动必须用达梦自己的驱动包。Spark 读取达梦的通用方式是通过 JDBCval df spark.read .format(jdbc) .option(url, jdbc:dm://192.168.1.100:5236/DAMENG) .option(dbtable, schema.orders) .option(user, test_user) .option(password, ***) .option(driver, dm.jdbc.driver.DmDriver) .load()有几个细节需要特别注意。第一驱动类名必须写对。达梦的 JDBC 驱动类是dm.jdbc.driver.DmDriver不是网上有些文章写的dm.jdbc.driver.DmDriver变了版本就随便改。驱动包要放到 Spark 的 classpath 里推荐放到$SPARK_HOME/jars目录或者提交作业时用--driver-class-path指定。第二分页读取性能。JDBC 读数据库默认是单分区读取数据量大后非常慢。需要通过分区参数让 Spark 并行读取.option(partitionColumn, order_id) .option(lowerBound, 1) .option(upperBound, 10000000) .option(numPartitions, 8)partitionColumn字段必须是数值类型Spark 会按这个字段的值范围把 SQL 拆成多段并行执行。如果表里没有合适的数值字段可以用row_number()窗口函数预先造一个自增列但这会加重数据库压力建议提前和 DBA 沟通一个低峰窗口。第三写回数据库时注意批量参数。Spark 默认通过 JDBC 一条条 insert效率很低。推荐这样配置df.write .mode(overwrite) .format(jdbc) .option(url, jdbc:dm://192.168.1.100:5236/DAMENG) .option(dbtable, schema.orders_result) .option(user, test_user) .option(password, ***) .option(driver, dm.jdbc.driver.DmDriver) .option(truncate, true) .option(batchsize, 1000) .save()mode(overwrite)配合truncatetrue会先清空再写入比 drop 表再建表快很多。batchsize控制在 500 到 2000 比较合适太大会导致数据库端内存上升太小则体现不出批量的效果。5.4 常见问题速查表把这一节提到的和以前项目里积累的高频问题整理成一张速查表方便大家排查时直接翻现象可能原因处理方式作业提交后卡在 Accepted 状态Yarn 队列资源不足查看队列负载释放资源或调大队列上限Executor 启动失败报 Unauthorized分配内存超过 Yarn container 限制调小 executor.memory 或 memoryOverhead任务很容易失败在 Shuffle 阶段Executor 数量太少 / 分区数不合理增加分区数检查动态分配配置广播变量超大数据导致 Executor 串压力大广播阈值不合理调小或调大 autoBroadcastJoinThreshold结合数据量判断开启 AQE 后结果和之前不一致自适应执行改变了 join 策略对比执行计划确认改动是否符合预期JDBC 读取数据库超时fetchsize 太小 / 单分区读太慢设置分区字段和 numPartitions调大 fetchsize达梦数据库中文乱码字符集不一致数据库连接串加兼容参数核对两端字符集6. 下一步从 CPU 到 GPU从批处理到更多场景聊完这些具体问题再回到一个更大的视角Scala 和 Spark 这套技术栈后续还能往哪些方向延伸。6.1 GPU 加速 Spark 的实际思路GPU 这个词在近两年的大数据领域出现频率越来越高。NVIDIA 的 RAPIDS 生态提供了一套基于 GPU 的 Spark 加速方案核心思路是让本来就跑在 JVM 里的算子尽可能在 GPU 上执行。举个最简单的例子原本需要几千万行数据做过滤、聚合、joinCPU 一个个处理换成 GPU 之后可以把一批数据整体并行计算处理时间能缩短一个数量级。但 GPU 加速在 Spark 场景下还远没有到开箱即用的程度。依赖 Spark 3.0 之后引入的 Resource Profile 机制需要在提交作业时声明每个 Task 需要的 GPU 数量同时还要考虑数据在 CPU 和 GPU 之间传输的开销。对大多数团队来说当前更落地的做法是把数据预处理、特征工程这类计算密集环节抽出来单独用 GPU 跑Spark 负责大规模数据编排。6.2 想进大厂做 Spark 开发这些知识点要能说出来很多读者问过 Spark 面试怎么准备。结合这些年面试别人的经验我筛了几个最常问的知识点都是真被问过很多次的Shuffle 的原理。数据在 map 端写磁盘、reduce 端拉取的全过程为什么会有溢写。宽依赖和窄依赖的区别以及它们在容错时的表现差异。RDD 的血统机制Stage 是怎么划分的。Spark 内存模型统一内存管理里执行内存和存储内存怎么互相借用。Catalyst 优化器和 Tungsten 执行引擎分别做了什么。AQEAdaptive Query Execution自适应查询执行在什么场景下能解决问题。这些问题光背八股没用必须结合源码和真实调优案例来说。比如问 AQE不能只说“动态优化 shuffle 分区数”最好能说出在数据倾斜场景下AQE 如何通过 coalesce 分区减少小文件又是怎么动态切换 join 策略的。6.3 这个系列后续可以怎么扩展第七篇的内容到这里主线基本上已经把 Scala 与 Spark 从环境到实战到调优串起来了。后续如果再写我想重点做两件事。一件是把这次案例里涉及的代码整理成一个完整的工程包含单元测试、打包脚本、提交模板放到网上供大家直接参考。另一件是写一期关于流批一体的内容把 Spark Streaming 和 Structured Streaming 在真实业务中的选型和落地经验分享出来。写在最后的个人体会踩过这么多坑之后我的体会其实是大数据分析项目的成败很多时候不是算法有多高级而是基础设施有没有理解透。CPU 核数上不去、内存频繁 OOM、数据源适配出问题任何一个都足以让一个看起来不复杂的作业跑不起来。别人可能花了很多时间在网上找答案但如果你把 Spark 的资源模型、内存模型和运行原理搞清楚了这些问题基本都能举一反三。最后再分享一个小技巧遇到任何 Spark 作业异常先别急着搜报错信息先去看 Spark UI 上的 Stages、Executors 和 SQL 页面。哪个阶段慢、哪个 Executor 挂了、Shuffle 读了多少数据这些信息比报错堆栈有用得多。把看 UI 变成习惯之后你会发现之前很多难以理解的问题其实答案就在眼前。