★ FEATURED ARTICLE 1. 为什么要在 Spring Boot 里塞一个 Spark很多做后端的朋友第一次听到Spring Boot 整合 Spark这个组合第一反应是这俩东西画风不太一样啊。Spring Boot 是典型的 Web 应用框架讲究的是快速启动、轻量、请求响应Spark 是分布式计算引擎讲究的是把一个大任务拆成无数小任务丢到集群上跑。把它们放一起本质上是要解决一个很实际的问题让业务系统能够通过一个 HTTP 接口触发一次大数据量的计算任务并且拿到结果或者至少拿到任务状态。这个需求在真实场景里非常常见。比如某个运营后台需要生成一份全量用户的行为分析报表数据量在千万级别用普通的 SQL 查询数据库会直接把库拖垮但如果用 Spark 跑一个批处理任务几分钟就能出结果。问题在于运营人员不可能去命令行敲spark-submit他们需要一个按钮点一下任务就跑起来跑完了页面上能看到结果。这时候 Spring Boot 就派上用场了——它负责接收 HTTP 请求、管理任务的生命周期、把 Spark 的计算能力包装成一个对前端友好的接口。所以这个整合的核心价值可以概括成三句话用 Spring Boot 做任务的门面和调度器用 Spark 做真正的计算引擎用 HTTP 接口做两者之间的桥梁。适合谁来参考呢如果你是一个 Java 后端开发者手头有 Spring Boot 的项目经验现在需要接入一些大数据处理能力但又不想单独维护一套 Spark 的提交脚本和调度系统那这套方案就是为你准备的。它不需要你成为 Spark 专家但需要你理解 Spark 的基本运行模型否则后面配置和调优的部分会看得很痛苦。我先把整体架构用文字描述清楚方便你建立全局观。整个系统分三层最上层是 Controller 层暴露 REST 接口接收任务参数中间是 Service 层负责构建 Spark 的配置对象、创建 SparkSession、提交任务、管理任务状态最下层是 Spark 的运行环境可以是本地的 local 模式也可以是 YARN 或者 Standalone 集群模式。这三层之间的边界要划清楚尤其是 SparkSession 的创建和销毁如果管理不当会出现内存泄漏或者任务互相干扰的问题这个后面会详细讲。还有一个容易被忽略的点Spark 的依赖包非常大。如果你用 Maven 引入spark-core和spark-sql打包出来的 fat jar 动辄两三百兆。这在开发阶段没什么感觉但到了部署阶段每次发版都要传这么大的包运维同学会想打人。所以后面我会专门讲依赖的 scope 怎么设置以及集群模式下怎么把依赖剥离出去。2. 依赖引入与项目骨架搭建2.1 Maven 依赖的取舍逻辑先看依赖。Spring Boot 的依赖不用多说用你熟悉的spring-boot-starter-web就行。关键是 Spark 相关的依赖怎么引。这里有一个非常重要的原则Spark 的依赖在开发阶段用compile在打包部署时尽量用provided。为什么因为集群环境上通常已经自带了 Spark 的运行时库。如果你把 Spark 的所有 jar 都打进自己的应用包里一来包体积巨大二来可能和集群上的 Spark 版本冲突导致各种诡异的ClassNotFoundException或者NoSuchMethodError。所以正确的做法是编译时需要这些类来写代码但运行时让集群提供。dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.4.1/version scopeprovided/scope /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.4.1/version scopeprovided/scope /dependency注意版本号里的_2.12是 Scala 的版本不是 Spark 的版本。Spark 3.x 默认用 Scala 2.12如果你用的 Spark 版本对应的是 2.13这里要改成_2.13。这个细节很多人第一次会搞错导致依赖拉不下来。但是如果你要在 local 模式下运行provided就会导致运行时找不到类。所以我的建议是开发环境用compile生产打包时通过 Maven Profile 切换成provided。这样本地调试方便部署时又不会把包撑大。2.2 项目目录结构的设计目录结构上我建议单独建一个spark包把所有和 Spark 相关的代码放在里面不要和普通的业务 Service 混在一起。原因很简单Spark 的代码有它自己的生命周期管理逻辑和普通的 CRUD 逻辑混在一起会让代码变得很难维护。com.example.demo ├── controller │ └── SparkJobController.java ├── service │ └── SparkJobService.java ├── spark │ ├── SparkSessionManager.java │ ├── JobConfig.java │ └── JobResult.java └── DemoApplication.javaSparkSessionManager是整个整合的核心类负责创建和管理 SparkSession。JobConfig是一个配置类用来承载任务的参数。JobResult是任务执行结果的封装。Controller 只负责接收请求和返回响应真正的逻辑在 Service 里。2.3 一个容易踩的坑Spring Boot 的类加载器这里要提前说一个坑。Spring Boot 用的是自己的类加载器LaunchedURLClassLoader而 Spark 在提交任务时也会涉及类加载。如果你在 local 模式下运行有时候会遇到ClassNotFoundException明明依赖已经引入了但就是找不到类。这通常是因为 Spark 的类加载器和 Spring Boot 的类加载器之间的委托关系出了问题。解决办法是在创建 SparkSession 的时候显式设置spark.driver.userClassPathFirst为true让 Spark 优先使用应用自己的类路径。这个参数在集群模式下也很有用可以避免集群上的旧版本 jar 覆盖你应用里的新版本。.config(spark.driver.userClassPathFirst, true)不过这个参数不是万能的有时候反而会引发新的冲突。我的经验是先不加遇到类加载问题再加加了之后如果问题更严重就换其他方案比如把冲突的依赖 shade 掉。3. SparkSession 的创建与生命周期管理3.1 为什么不能每次请求都创建一个 SparkSession这是新手最容易犯的错误。很多人写代码的时候习惯性地在每次请求进来的时候SparkSession.builder().getOrCreate()觉得getOrCreate会帮你复用。但实际上在 Spring Boot 的 Web 环境里每次请求可能由不同的线程处理而 SparkSession 本身是线程安全的但它的创建过程是有开销的。更重要的是SparkSession 持有一个 SparkContext而 SparkContext 一旦启动就会占用一堆资源driver 的内存、和集群的通信连接、executor 的分配等等。如果你每次请求都创建一个新的很快就会发现内存爆了或者集群上的 executor 被占满了。所以正确的做法是整个应用生命周期内只创建一个 SparkSession所有任务共享它。用 Spring 的Bean和单例模式来管理。Configuration public class SparkSessionManager { private SparkSession sparkSession; PostConstruct public void init() { sparkSession SparkSession.builder() .appName(spring-boot-spark-demo) .master(local[*]) .config(spark.sql.shuffle.partitions, 10) .config(spark.driver.memory, 2g) .getOrCreate(); } public SparkSession getSession() { return sparkSession; } PreDestroy public void destroy() { if (sparkSession ! null) { sparkSession.stop(); } } }注意PreDestroy里的stop()调用。这个非常重要如果不调用应用关闭的时候 SparkContext 可能不会正常释放导致端口被占用或者资源泄漏。我在本地调试的时候就遇到过改完代码重启应用结果报端口冲突就是因为上一次的 SparkContext 没关干净。3.2 local 模式和集群模式的配置差异master这个参数决定了 Spark 跑在哪里。local[*]表示用本地所有 CPU 核心跑local[2]表示用两个核心。这种模式适合开发和测试不需要任何集群环境。如果要跑在集群上master要改成对应的集群地址。比如 YARN 模式是yarnStandalone 模式是spark://host:7077。但这里有个关键点在集群模式下你不能在代码里硬编码 master 地址因为不同环境的集群地址不一样。正确的做法是通过配置文件或者环境变量注入。Value(${spark.master:local[*]}) private String sparkMaster;这样在本地开发时用默认的local[*]部署到集群时通过配置覆盖成yarn。这个设计看起来简单但实际项目中很多人会忽略导致代码在本地跑得好好的一上集群就挂。还有一个参数spark.submit.deployMode在 YARN 模式下可以设置为client或者cluster。client模式下 driver 跑在提交任务的机器上cluster模式下 driver 跑在集群的某个节点上。对于 Spring Boot 应用来说通常用client模式因为 driver 需要和 Web 应用在一起才能把结果返回给 HTTP 请求。3.3 资源参数的设置与计算Spark 的资源参数直接决定了任务能跑多快、能处理多大数据量。最核心的三个参数是spark.driver.memory、spark.executor.memory和spark.executor.cores。spark.driver.memory控制 driver 的内存。driver 负责协调任务、收集结果如果你的任务需要把大量数据 collect 回 driver这个值要设大一点。一般 2g 到 4g 够用但如果你的任务会返回几十万行数据就要慎重了因为 collect 操作会把所有数据拉到 driver 内存里很容易 OOM。spark.executor.memory控制每个 executor 的内存。这个值的计算有个经验公式每个 executor 的内存 总内存 / executor 数量。比如集群有 100g 内存你想开 10 个 executor那每个就是 10g。但要注意executor 内存里有一部分是给 Spark 自己用的比如 shuffle 的缓冲区实际能用来缓存数据的只有一部分。通常建议每个 executor 的内存不低于 4g否则稍微大一点的数据集就会 spill 到磁盘。spark.executor.cores控制每个 executor 的 CPU 核心数。这个值一般设 2 到 5 之间。设太小executor 的并行度不够设太大一个 executor 上跑的 task 太多GC 压力会很大。我个人的经验是设 4 比较均衡。.config(spark.executor.memory, 4g) .config(spark.executor.cores, 4) .config(spark.executor.instances, 10)spark.executor.instances是 executor 的数量。这个值不是越大越好因为每个 executor 都要和 driver 通信数量太多会让 driver 成为瓶颈。一般来说几十个 executor 已经能处理很大的数据量了。4. HTTP 接口触发 Spark 任务的完整实现4.1 Controller 层的设计Controller 层的设计要遵循一个原则接口要简单参数要清晰返回要明确。不要让前端传一堆 Spark 的配置参数过来那样既不安全也不好维护。正确的做法是定义几个业务层面的参数比如任务类型、数据范围、输出格式然后在 Service 层把这些业务参数翻译成 Spark 的配置。RestController RequestMapping(/api/spark) public class SparkJobController { Autowired private SparkJobService sparkJobService; PostMapping(/submit) public ResponseEntityJobResult submitJob(RequestBody JobRequest request) { JobResult result sparkJobService.executeJob(request); return ResponseEntity.ok(result); } GetMapping(/status/{jobId}) public ResponseEntityJobStatus getStatus(PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getJobStatus(jobId)); } }这里我设计了两个接口一个提交任务一个查询状态。为什么要分开因为 Spark 任务通常是异步的提交之后不会立刻返回结果。如果做成同步的HTTP 请求会一直挂着直到任务跑完这期间如果任务跑了十分钟前端就超时了。所以更好的做法是提交任务后立刻返回一个 jobId前端拿着 jobId 轮询状态。4.2 异步执行与任务状态管理异步执行的核心是把任务丢到一个线程池里跑然后立刻返回。这里要注意Spark 的任务提交本身是异步的但如果你在提交后调用collect()或者count()这种 action 操作就会阻塞当前线程。所以要把整个任务执行逻辑放到一个独立的线程里。Service public class SparkJobService { private final ExecutorService executorService Executors.newFixedThreadPool(5); private final MapString, JobStatus jobStatusMap new ConcurrentHashMap(); public JobResult executeJob(JobRequest request) { String jobId UUID.randomUUID().toString(); jobStatusMap.put(jobId, JobStatus.running()); executorService.submit(() - { try { // 真正的 Spark 任务逻辑 DatasetRow result runSparkJob(request); long count result.count(); jobStatusMap.put(jobId, JobStatus.success(count)); } catch (Exception e) { jobStatusMap.put(jobId, JobStatus.failed(e.getMessage())); } }); return new JobResult(jobId, submitted); } }线程池的大小要根据你的集群资源来定。如果集群只能同时跑 5 个任务线程池设成 10 也没用多出来的任务只会排队等待。而且要注意Spark 任务之间是会互相抢资源的如果同时提交太多任务每个任务分到的 executor 就少了反而都跑得慢。4.3 任务结果的返回方式任务结果的返回有三种常见方式各有适用场景。第一种是直接返回小结果集。如果任务的结果只有几百行可以直接collect()回来转成 JSON 返回给前端。这种方式最简单但只适合小数据量。第二种是返回统计值。如果前端只需要知道总数、最大值、平均值这些可以在 Spark 里算好只返回几个数字。这种方式对内存最友好。第三种是写入外部存储后返回路径。如果结果有几十万行不适合直接返回可以把结果写成 Parquet 或者 CSV 文件存到 HDFS 或者对象存储上然后返回一个下载链接。这种方式最适合大数据量的场景。// 方式一直接返回 ListRow rows result.collectAsList(); // 方式二返回统计值 long total result.count(); double avg result.select(avg(amount)).first().getDouble(0); // 方式三写入外部存储 result.write().mode(overwrite).parquet(hdfs://path/to/output);我个人的建议是默认用方式三除非你确定结果很小。因为方式一和方式二都有 driver OOM 的风险而方式三把压力分散到了 executor 上更安全。5. Spark 配置优化的实战经验5.1 序列化器的选择Spark 默认用的是 Java 序列化器性能一般。换成 Kryo 序列化器可以显著提升性能尤其是在 shuffle 数据量大的时候。Kryo 的序列化速度比 Java 快好几倍序列化后的数据也更小。.config(spark.serializer, org.apache.spark.serializer.KryoSerializer) .config(spark.kryo.registrationRequired, false)但 Kryo 有个坑它需要你注册自定义的类否则性能会打折扣。如果你不注册Kryo 会把每个类的全名都写进序列化数据里反而更占空间。所以如果你的任务里用了自定义的 POJO最好注册一下。.config(spark.kryo.classesToRegister, com.example.demo.UserBehavior,com.example.demo.OrderInfo)5.2 shuffle 分区数的调整spark.sql.shuffle.partitions这个参数默认是 200。这个默认值对于小数据量来说太大了会导致大量的小任务每个任务处理的数据量很少调度开销反而成了瓶颈。对于中等数据量几 GB设成 50 到 100 比较合适。对于小数据量几百 MB设成 10 到 20 就够了。.config(spark.sql.shuffle.partitions, 50)怎么判断设多少合适一个经验法则是每个分区的数据量在 100MB 到 200MB 之间比较理想。如果你知道总数据量是 5GB那设 50 个分区每个分区大约 100MB就比较合适。5.3 内存管理的调优Spark 的内存管理分两块execution 内存和 storage 内存。execution 内存用于 shuffle、join、sort 等操作storage 内存用于缓存数据。默认情况下两者各占 50%但在某些场景下需要调整。如果你的任务有很多 shuffle 操作可以调大 execution 内存的比例.config(spark.memory.fraction, 0.8) .config(spark.memory.storageFraction, 0.3)spark.memory.fraction控制的是 Spark 内存占总内存的比例默认 0.6。调大到 0.8 意味着给 Spark 更多内存留给用户对象的内存就少了。spark.memory.storageFraction控制的是 storage 内存占 Spark 内存的比例默认 0.5。调小到 0.3 意味着更多内存给 execution 用。这两个参数需要根据实际任务的特点来调。如果任务经常 OOM先调大spark.memory.fraction如果任务经常 spill 到磁盘调小spark.memory.storageFraction。5.4 数据倾斜的处理数据倾斜是 Spark 任务最常见的性能杀手。表现是大部分 task 很快就跑完了但有几个 task 跑得特别慢甚至卡住不动。原因通常是某个 key 的数据量特别大导致处理这个 key 的 task 成了瓶颈。处理数据倾斜有几种常见方法。第一种是加盐打散给倾斜的 key 加上随机前缀让它分散到多个 task 上最后再去掉前缀聚合。第二种是广播小表如果 join 的一边是小表可以用 broadcast join 避免 shuffle。第三种是拆分大 key把倾斜的 key 单独拿出来处理剩下的正常处理。// 广播小表 DatasetRow result largeTable.join( broadcast(smallTable), join_key );广播 join 的原理是把小表复制到每个 executor 上这样就不需要 shuffle 大表了。但小表不能太大一般建议不超过几百 MB否则 driver 和 executor 的内存压力会很大。6. 集群模式下的部署与排错6.1 打包与提交的注意事项集群模式下你的 Spring Boot 应用需要打包成一个可执行的 jar然后提交到集群上运行。但这里有个问题Spark 的依赖不能打进包里否则会和集群上的 Spark 冲突。所以打包的时候要用providedscope或者用 shade 插件把 Spark 的类排除掉。plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId configuration artifactSet excludes excludeorg.apache.spark:*/exclude excludeorg.scala-lang:*/exclude /excludes /artifactSet /configuration /plugin提交任务的时候用spark-submit命令指定你的 jar 包和主类。但注意如果你的应用是一个 Web 服务它需要一直运行而不是跑完就退出。这时候spark-submit的用法就和普通的批处理任务不一样了。一种做法是用spark-submit提交一个长期运行的服务但这种方式不太优雅因为 Spark 的提交机制本来是为批处理设计的。另一种做法是把 Spring Boot 应用独立部署然后在代码里通过SparkSession.builder().master(yarn)连接到集群。这种方式更灵活但需要你的应用能访问到集群的配置。6.2 常见的类冲突与排查方法类冲突是集群模式下最常见的问题。表现是本地跑得好好的一上集群就报NoSuchMethodError或者ClassNotFoundException。原因通常是集群上的某个 jar 版本和你应用里的版本不一致。排查方法分三步。第一步看异常堆栈找到是哪个类出的问题。第二步用mvn dependency:tree看你的应用里这个类来自哪个 jar。第三步去集群上确认这个 jar 的版本。如果版本不一致要么升级你的依赖要么把冲突的类 shade 掉。mvn dependency:tree | grep spark还有一个工具叫jar tf可以看 jar 包里有哪些类。如果你怀疑某个类被重复打包了可以用这个命令确认。6.3 日志与监控的接入集群模式下日志不会直接输出到你的控制台而是分散在各个 executor 的节点上。所以你需要一套日志收集方案。最简单的做法是在 Spark 配置里设置日志级别把日志写到 HDFS 上。.config(spark.eventLog.enabled, true) .config(spark.eventLog.dir, hdfs://path/to/spark-logs)然后通过 Spark 的 Web UI 查看任务执行情况。Web UI 默认在 4040 端口可以看到每个 stage 的耗时、每个 task 的数据量、shuffle 的读写量等信息。这些信息对于调优非常有用。如果你想把日志接入到现有的监控系统可以在应用里用 Logback 或者 Log4j2 配置日志输出把 Spark 的日志和应用的日志统一收集。但要注意Spark 的日志量很大如果全部收集存储成本会很高。建议只收集 WARN 和 ERROR 级别的日志。7. 几个我踩过的坑和对应的解法7.1 SparkSession 重复创建导致端口冲突前面提到过SparkSession 要单例。但即使你用了单例如果在测试的时候频繁重启应用可能会遇到 4040 端口被占用的问题。这是因为上一次的 SparkContext 没有完全关闭端口还没释放。解决办法是在PreDestroy里调用sparkSession.stop()并且确保stop()被调用。如果还是不行可以在启动时检查端口是否被占用如果被占用就换一个端口。.config(spark.ui.port, 4041)7.2 collect 操作导致 driver OOM这个坑我踩过好几次。任务在 executor 上跑得好好的一到 collect 就 OOM。原因是 collect 会把所有数据拉到 driver 的内存里如果数据量大driver 的内存根本不够。解决办法是尽量不要用 collect。如果一定要用先估算数据量。比如你知道结果有 10 万行每行 1KB那就是 100MBdriver 内存设 2g 是够的。但如果结果有 1000 万行那就别 collect 了直接写文件。7.3 任务卡在某个 stage 不动这种情况通常是数据倾斜或者资源不足。先看 Spark UI找到卡住的 stage看它的 task 分布。如果大部分 task 都完成了只有几个没完成那就是数据倾斜。如果所有 task 都很慢那就是资源不足需要加 executor 或者加内存。还有一种可能是 shuffle 的文件太多导致磁盘 IO 成了瓶颈。这时候可以调大spark.sql.shuffle.partitions减少每个分区的文件数。7.4 序列化异常Task not serializable是 Spark 里最常见的异常之一。原因是你在闭包里引用了一个不可序列化的对象比如数据库连接、Spring 的 Bean 等。解决办法是在闭包里只引用可序列化的对象。如果必须引用一个不可序列化的对象可以把它声明为transient然后在 executor 端重新初始化。public class MyJob implements Serializable { private transient DatabaseConnection conn; public void run() { // 在 executor 端重新初始化 if (conn null) { conn createConnection(); } } }这个坑的本质是Spark 会把闭包序列化后发送到 executor 上执行如果闭包里引用了 driver 端的对象那个对象必须能被序列化。Spring 的 Bean 通常不能被序列化所以不要在 Spark 的闭包里直接引用Autowired的 Bean。8. 一些实用的配置模板和参数速查8.1 local 模式的最小配置SparkSession.builder() .appName(local-job) .master(local[*]) .config(spark.sql.shuffle.partitions, 10) .config(spark.driver.memory, 2g) .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) .getOrCreate();这套配置适合本地开发和测试启动快资源占用少。8.2 集群模式的推荐配置SparkSession.builder() .appName(cluster-job) .master(yarn) .config(spark.submit.deployMode, client) .config(spark.executor.memory, 4g) .config(spark.executor.cores, 4) .config(spark.executor.instances, 10) .config(spark.sql.shuffle.partitions, 100) .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) .config(spark.driver.userClassPathFirst, true) .getOrCreate();这套配置适合中等规模的数据处理任务可以根据实际情况调整 executor 的数量和内存。8.3 关键参数速查表参数默认值建议值作用spark.sql.shuffle.partitions20010-100shuffle 后的分区数spark.executor.memory1g4g-8g每个 executor 的内存spark.executor.cores12-5每个 executor 的 CPU 核心数spark.executor.instances2根据集群定executor 的数量spark.serializerJavaKryo序列化器spark.memory.fraction0.60.6-0.8Spark 内存占总内存比例spark.memory.storageFraction0.50.3-0.5storage 内存占 Spark 内存比例这张表可以贴在显示器旁边调参的时候随时参考。但记住没有一套参数是万能的每个任务的数据量、计算逻辑、集群环境都不一样最好的参数是测出来的不是抄出来的。8.4 一个完整的任务执行示例最后给一个完整的示例把前面的内容串起来。这个示例接收一个 HTTP 请求读取一个 Parquet 文件做一个 group by 聚合然后把结果写回 Parquet。Service public class SparkJobService { Autowired private SparkSessionManager sessionManager; public JobResult executeJob(JobRequest request) { String jobId UUID.randomUUID().toString(); executorService.submit(() - { try { SparkSession spark sessionManager.getSession(); DatasetRow input spark.read() .parquet(request.getInputPath()); DatasetRow result input .groupBy(category) .agg(sum(amount).as(total_amount)); result.write() .mode(overwrite) .parquet(request.getOutputPath()); jobStatusMap.put(jobId, JobStatus.success()); } catch (Exception e) { jobStatusMap.put(jobId, JobStatus.failed(e.getMessage())); } }); return new JobResult(jobId, submitted); } }这个示例虽然简单但涵盖了整合的核心流程接收请求、获取 SparkSession、执行计算、写入结果、更新状态。你可以在这个基础上根据实际需求扩展比如增加参数校验、增加重试逻辑、增加任务超时处理等。我在实际项目里用这套方案跑了半年多处理了上千个任务整体稳定性还不错。最大的体会是Spark 的配置调优是一个持续的过程不要指望一次调好就一劳永逸。随着数据量的增长和业务逻辑的变化之前合适的参数可能就不合适了需要定期 review 和调整。另外日志和监控一定要做好否则出了问题只能靠猜效率会非常低。
阅读完成 · 觉得有帮助？
