1. 项目缘起与整体架构思路把 Spark 塞进 Spring Boot 里再用 HTTP 接口去触发任务这个需求我第一次碰到是在一个数据清洗与指标计算的后台系统里。当时的场景很典型业务侧有一堆 CSV 和日志文件需要做聚合统计算法同事用 Spark 写好了逻辑但运维和前端不想碰命令行希望点个按钮就能跑任务、看进度、拿结果。于是就有了这套“Spring Boot Spark HTTP 调用”的组合。先说清楚这个项目到底解决什么问题。传统 Spark 作业的提交方式是spark-submit脚本化、命令行化适合调度平台但对业务系统不友好。而 Spring Boot 天生适合做 Web 服务把 Spark 的SparkSession生命周期交给 Spring 容器管理对外暴露 REST 接口就能实现“请求即任务”的效果。这套方案适合谁适合有一定 Java 基础、想快速把离线计算能力嵌入到业务系统里的后端开发者也适合做数据中台、报表平台、批处理调度这类项目的同学。整体架构上我采用的是分层设计。最上层是 Controller负责接收 HTTP 请求、校验参数、返回任务 ID中间是 Service 层封装 Spark 任务的提交、状态查询和结果获取底层是 Spark 的SparkSession和具体的业务计算逻辑。这里有个关键决策点SparkSession是单例还是每次请求新建我的经验是在 local 模式下可以做成单例常驻因为 local 模式本身就是本地 JVM 内的线程池反复创建销毁开销大而在集群模式下更推荐每次任务独立提交避免资源长期占用和状态污染。为什么这么设计因为 Spark 的SparkContext一旦停止就无法重启单例模式下如果某个任务异常导致 context 关闭整个服务就废了。所以我在实际项目里做了一个折中local 模式用单例加任务队列集群模式用SparkLauncher或 REST 提交到集群Spring Boot 只做任务管理和状态轮询。这样既保证了本地调试的便利性又兼顾了生产环境的稳定性。还有一个容易被忽略的点是依赖冲突。Spring Boot 自带的日志、Jackson、Netty 等组件和 Spark 的依赖经常打架尤其是slf4j和log4j的版本问题。我的做法是在pom.xml里显式排除 Spark 自带的日志实现统一用 Spring Boot 的日志体系同时在打包时用maven-shade-plugin做重定位避免类加载冲突。这一步不做后面启动时报NoSuchMethodError会让你怀疑人生。2. 环境搭建与依赖配置的实操细节2.1 版本选型与兼容性对照版本选型是这个项目的第一道坎。Spring Boot 和 Spark 的版本不是随便配的Scala 版本、JDK 版本、Hadoop 版本之间有一张隐形的兼容表。我踩过的坑是Spring Boot 2.7 配 Spark 3.2 时JDK 必须用 8 或 11用 17 会遇到IllegalAccessError因为 Spark 内部大量使用了反射访问 JDK 内部类。下面这张表是我实测下来比较稳的组合Spring Boot 版本Spark 版本Scala 版本JDK 版本备注2.7.x3.2.x2.128 / 11最稳定推荐2.7.x3.3.x2.128 / 11需注意 Jackson 冲突3.0.x3.4.x2.1211 / 17需加 JVM 参数开放模块2.6.x3.1.x2.128老项目维护用选 Scala 2.12 而不是 2.13是因为大部分生产环境的 Hadoop 生态还是 2.12 为主2.13 的包相对少。JDK 方面如果你非要用 17记得在启动参数里加上--add-opens java.base/java.langALL-UNNAMED这类开放模块的配置否则 Spark 的序列化会直接报错。2.2 Maven 依赖的取舍与排除依赖配置我直接给一份可用的骨架重点看exclusions部分dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.2.4/version exclusions exclusion groupIdorg.slf4j/groupId artifactIdslf4j-log4j12/artifactId /exclusion exclusion groupIdlog4j/groupId artifactIdlog4j/artifactId /exclusion exclusion groupIdorg.codehaus.janino/groupId artifactIdjanino/artifactId /exclusion /exclusions /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.2.4/version /dependency为什么要排除janino因为 Spring Boot 的spring-boot-devtools或某些模板引擎也会引入 janino版本不一致时会导致 Spark SQL 的代码生成失败。排除后由 Spring Boot 统一管理版本问题就消失了。另外spark-sql是必须的因为实际业务里没人只用 RDDDataFrame 和 Dataset 才是主流。如果你还要用 Spark Streaming再加spark-streaming但注意流式任务和 Web 服务的生命周期管理会更复杂建议单独拆服务。2.3 配置文件的关键参数application.yml里我一般会放这些配置spark: app-name: spring-boot-spark-demo master: local[*] driver-memory: 2g executor-memory: 2g executor-cores: 2 sql-shuffle-partitions: 200 serializer: org.apache.spark.serializer.KryoSerializer kryo-registrator: com.example.MyKryoRegistratormaster在 local 模式下写local[*]表示用所有可用核心调试时可以用local[2]限制核心数方便观察资源占用。sql-shuffle-partitions默认是 200小数据量下这个值太大会产生大量小文件我一般会根据数据量调到 20 到 50 之间。序列化器强烈建议用 Kryo比 Java 原生序列化快好几倍但记得注册自定义类否则性能提升有限。提示driver-memory在 local 模式下其实就是 JVM 堆内存设置时不要超过物理内存的 70%否则容易触发 OOM Killer。3. SparkSession 的 Spring 化管理与 HTTP 接口设计3.1 SparkSession 单例 Bean 的正确写法把SparkSession交给 Spring 管理核心是写一个Configuration类Configuration public class SparkConfig { Value(${spark.app-name}) private String appName; Value(${spark.master}) private String master; Bean(destroyMethod stop) public SparkSession sparkSession() { return SparkSession.builder() .appName(appName) .master(master) .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) .config(spark.sql.shuffle.partitions, 50) .getOrCreate(); } }这里destroyMethod stop很关键Spring 容器关闭时会自动调用sparkSession.stop()避免 JVM 无法退出。我见过有人忘了加这个结果服务停了但 Spark 的线程还在跑进程一直挂着。但要注意SparkSession是重量级对象创建一次要几秒钟。如果你的服务是多实例部署每个实例都会创建一个 SparkSessionlocal 模式下没问题集群模式下会各自连集群资源消耗要提前评估。3.2 任务提交接口的设计HTTP 接口我设计成三个提交任务、查询状态、获取结果。提交接口用 POST返回一个任务 IDRestController RequestMapping(/api/spark) public class SparkJobController { Autowired private SparkJobService sparkJobService; PostMapping(/submit) public ResponseEntityString submit(RequestBody JobRequest request) { String jobId sparkJobService.submitJob(request); return ResponseEntity.ok(jobId); } GetMapping(/status/{jobId}) public ResponseEntityJobStatus status(PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getStatus(jobId)); } GetMapping(/result/{jobId}) public ResponseEntityListMapString, Object result(PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getResult(jobId)); } }为什么用异步提交而不是同步等待因为 Spark 任务可能跑几分钟甚至几小时HTTP 连接扛不住。我的做法是用ExecutorService或者CompletableFuture异步执行任务状态存在ConcurrentHashMap里前端轮询状态接口。任务 ID 用 UUID简单可靠。JobRequest里一般包含数据源路径、输出路径、计算类型、过滤条件等参数。这里有个安全考虑绝对不要让用户直接传 SQL 或代码片段否则就是注入漏洞。我的做法是预定义几种计算模板用户只传参数模板在服务端映射成具体的 DataFrame 操作。3.3 任务状态管理与结果缓存状态管理我用一个简单的枚举PENDING、RUNNING、SUCCESS、FAILED。每个任务对应一个JobContext对象包含状态、开始时间、结束时间、错误信息、结果集。结果集我一般限制条数比如最多返回 1000 行避免内存爆掉。public class JobContext { private String jobId; private volatile JobStatus status; private long startTime; private long endTime; private String errorMsg; private ListMapString, Object result; }结果缓存用ConcurrentHashMap就够了但如果任务量大建议加过期清理策略比如任务完成 1 小时后自动移除。我试过用 Caffeine 做本地缓存效果不错配置expireAfterWrite(1, TimeUnit.HOURS)即可。注意结果集不要直接存 DataFrameDataFrame 持有 Spark 的 lineage 和资源引用存多了会导致内存泄漏。一定要collectAsList()转成普通 Java 集合再存。4. 集群模式与 local 模式的切换与优化4.1 两种模式的本质区别local 模式和集群模式的区别不只是master参数不同。local 模式下driver 和 executor 在同一个 JVM 里共享内存和 CPU调试方便但资源隔离差。集群模式下driver 在 Spring Boot 服务里executor 在集群的 worker 节点上通过网络通信资源隔离好但延迟高。我一般这样选开发和测试用 local生产环境用集群。但有些场景下数据量不大、计算逻辑简单local 模式反而更快因为省去了网络传输和任务调度开销。判断标准是数据量超过单机内存的 50%或者计算需要多节点并行就上集群。集群模式下Spring Boot 服务本身不跑计算只做任务提交和状态管理。提交方式有两种一种是用SparkLauncher在本地启动一个提交进程另一种是直接调集群的 REST 接口。SparkLauncher的好处是能拿到任务句柄方便监控坏处是依赖本地有 Spark 客户端环境。SparkAppHandle handle new SparkLauncher() .setAppResource(/path/to/job.jar) .setMainClass(com.example.SparkJob) .setMaster(yarn) .setDeployMode(cluster) .addAppArgs(jobId, inputPath, outputPath) .startApplication();这段代码在 local 模式下也能跑但setMaster(yarn)需要 Hadoop 配置。如果你的集群是 Standalone 模式改成spark://host:7077即可。4.2 资源配置的计算过程资源配置是优化的核心。假设你的集群有 3 个节点每个节点 16 核 64G 内存那么总资源是 48 核 192G。一般建议 executor 数量不超过总核数的 80%留一部分给系统和其他服务。我常用的配置公式是executor 数量 总核数 / 每个 executor 的核数每个 executor 的核数建议 2 到 5太多会导致 HDFS IO 竞争executor 内存 节点内存 / 每节点 executor 数量 * 0.8比如 3 节点 16 核 64G我配置 6 个 executor每个 4 核 16G这样用了 24 核 96G留了一半资源给系统。driver-memory一般 2G 到 4G 足够除非你要collect大量数据到 driver。spark: executor: instances: 6 cores: 4 memory: 16g driver: memory: 4g这些参数在 local 模式下大部分无效因为 local 模式只有一个 JVM。所以我在代码里做了判断如果是 local 模式忽略 executor 相关配置只设置 driver 内存。4.3 性能优化的几个关键点第一个优化点是分区数。spark.sql.shuffle.partitions默认 200小数据量下会产生大量小任务调度开销比计算还大。我的经验是数据量小于 1G 时设 10 到 201G 到 10G 设 50 到 10010G 以上再考虑 200 以上。第二个优化点是缓存策略。如果同一个 DataFrame 被多次使用一定要cache()或persist()。但不要滥用缓存会占内存用完后unpersist()释放。我见过有人把所有 DataFrame 都 cache结果 executor OOM。第三个优化点是广播变量。小表 join 大表时用broadcast()把小表广播出去避免 shuffle。这个优化效果非常明显我实测过一个 10G 大表 join 100M 小表的场景广播后耗时从 8 分钟降到 1 分钟。DatasetRow result largeTable.join( broadcast(smallTable), joinKey );第四个优化点是序列化。Kryo 序列化比 Java 原生快 2 到 5 倍但需要注册类。我一般会写一个KryoRegistrator把所有自定义的 POJO 都注册进去。提示优化不是一次性的要结合 Spark UI 看每个 stage 的耗时和 shuffle 数据量针对性调整。盲目调参不如先看监控。5. 常见问题排查与避坑经验5.1 启动阶段的典型报错问题一java.lang.NoClassDefFoundError: org/slf4j/Logger这是日志冲突的典型表现。原因是 Spark 自带的 slf4j 版本和 Spring Boot 的不一致。解决办法是在pom.xml里排除 Spark 的slf4j-log4j12统一用 Spring Boot 的spring-boot-starter-logging。问题二java.lang.IllegalAccessError: class org.apache.spark... cannot access class sun.nio.ch...这是 JDK 17 的模块化限制导致的。解决办法是加 JVM 参数--add-opens java.base/sun.nio.chALL-UNNAMED --add-opens java.base/java.langALL-UNNAMED --add-opens java.base/java.utilALL-UNNAMED问题三org.apache.spark.SparkException: A master URL must be set说明SparkSession创建时没有设置 master。检查配置文件是否被正确加载或者代码里是否硬编码了master。5.2 运行阶段的状态异常问题四任务一直卡在 RUNNING 状态可能原因有三个一是数据倾斜某个 key 的数据量特别大二是资源不足executor 申请不到三是死锁比如在 Spark 任务里调用了 Web 服务的同步接口。排查方法是看 Spark UI 的 stage 详情找到卡住的 task。问题五OutOfMemoryError: Java heap spacedriver OOM 一般是collect()数据量太大改成take()或写文件。executor OOM 一般是分区数据不均匀或缓存太多调整分区数或减少缓存。问题六HTTP 请求超时Spark 任务执行时间超过网关的超时限制。解决办法是改成异步提交前端轮询状态。如果必须同步把网关超时调到 10 分钟以上但不推荐。5.3 常见问题速查表问题现象可能原因排查方法解决方案启动报 NoClassDefFoundError依赖冲突mvn dependency:tree排除冲突包JDK 17 反射报错模块化限制看报错类名加 add-opens 参数任务卡住不动数据倾斜/资源不足Spark UI 看 stage加盐/调资源driver OOMcollect 数据过多看 driver 日志改 take 或写文件executor OOM缓存过多/分区不均看 executor 日志减缓存/调分区结果返回慢结果集太大看接口耗时限制返回条数服务无法退出SparkSession 未关闭jstack 看线程加 destroyMethod5.4 独家避坑技巧第一个技巧在 Spring Boot 启动类上加EnableAsync把 Spark 任务放到独立线程池。不要用默认的SimpleAsyncTaskExecutor那个每次请求都新建线程任务多了会爆。自定义一个ThreadPoolTaskExecutor核心线程数设 5 到 10队列容量设 100。第二个技巧给 Spark 任务加超时控制。用Future.get(timeout, TimeUnit)包一层超时后调用future.cancel(true)并记录日志。但注意 Spark 任务不一定能立即中断所以还要配合sparkSession.stop()或SparkContext.cancelJobGroup()。第三个技巧日志分级。Spark 的日志非常啰嗦默认 INFO 级别会刷屏。在logback-spring.xml里把org.apache.spark设为 WARN只保留关键信息。调试时再临时开 DEBUG。logger nameorg.apache.spark levelWARN/ logger nameorg.apache.hadoop levelWARN/第四个技巧本地模式下的临时目录。Spark 会在/tmp下创建大量临时文件时间长了会占满磁盘。在spark-defaults.conf或代码里设置spark.local.dir到一个大容量目录并定期清理。.config(spark.local.dir, /data/spark-tmp)第五个技巧任务幂等性。同一个任务重复提交时要保证结果一致。我的做法是输出路径带任务 ID或者写入前先删除旧数据。否则重跑任务会报“路径已存在”的错误。6. 完整实操流程与代码骨架6.1 从零搭建的步骤清单第一步创建 Spring Boot 项目选 Maven、JDK 8 或 11引入spring-boot-starter-web和spark-core、spark-sql。第二步配置pom.xml排除日志冲突加maven-shade-plugin做打包。第三步写SparkConfig配置类创建SparkSessionBean。第四步写SparkJobService实现任务提交、状态管理、结果缓存。第五步写SparkJobController暴露三个 HTTP 接口。第六步写具体的 Spark 计算逻辑比如 CSV 读取、聚合、写回。第七步本地用local[*]测试通过后改成集群模式部署。6.2 核心 Service 的代码实现Service public class SparkJobService { Autowired private SparkSession sparkSession; private final MapString, JobContext jobMap new ConcurrentHashMap(); private final ExecutorService executor Executors.newFixedThreadPool(5); public String submitJob(JobRequest request) { String jobId UUID.randomUUID().toString(); JobContext context new JobContext(jobId); jobMap.put(jobId, context); executor.submit(() - { context.setStatus(JobStatus.RUNNING); context.setStartTime(System.currentTimeMillis()); try { ListMapString, Object result executeJob(request); context.setResult(result); context.setStatus(JobStatus.SUCCESS); } catch (Exception e) { context.setErrorMsg(e.getMessage()); context.setStatus(JobStatus.FAILED); } finally { context.setEndTime(System.currentTimeMillis()); } }); return jobId; } private ListMapString, Object executeJob(JobRequest request) { DatasetRow df sparkSession.read() .option(header, true) .csv(request.getInputPath()); DatasetRow result df.groupBy(category) .agg(functions.sum(amount).as(total)); return result.limit(1000).collectAsList().stream() .map(row - row.getValuesMap(row.schema().fieldNames())) .collect(Collectors.toList()); } }这段代码是骨架实际业务里executeJob会根据request.getJobType()走不同的分支。我一般用策略模式每种任务类型对应一个JobHandler实现类Service 里根据类型路由。6.3 集群模式下的提交改造集群模式下executeJob不再直接跑计算而是提交到集群。改造后的逻辑private ListMapString, Object executeJob(JobRequest request) throws Exception { SparkAppHandle handle new SparkLauncher() .setAppResource(request.getJarPath()) .setMainClass(request.getMainClass()) .setMaster(request.getMaster()) .setDeployMode(cluster) .addAppArgs(request.getInputPath(), request.getOutputPath()) .startApplication(); while (!handle.getState().isFinal()) { Thread.sleep(2000); } if (handle.getState() SparkAppHandle.State.FINISHED) { return readResultFromOutput(request.getOutputPath()); } else { throw new RuntimeException(Spark job failed: handle.getState()); } }这里的结果读取是从输出路径读文件而不是从 Spark 直接返回。因为集群模式下 driver 和 executor 分离结果一般写 HDFS 或对象存储再由 Spring Boot 服务读取。6.4 参数调优的实测记录我拿一个 5G 的 CSV 文件做过对比测试任务是按类别聚合金额。测试环境是 3 节点每节点 8 核 32G。配置项默认值调优值耗时对比shuffle partitions200504分20秒 → 2分50秒序列化器JavaKryo2分50秒 → 2分10秒executor 内存1g4g频繁 GC → 稳定广播 join未用使用小表 join 快 3 倍调优后总耗时从 6 分钟降到 2 分半效果还是很明显的。但要注意调优不是越多越好比如 executor 内存设太大会导致 GC 时间变长反而变慢。我的经验是先看 Spark UI 找瓶颈再针对性调参每次只改一个参数记录对比结果。7. 个人实操体会与后续扩展方向这套方案我在两个项目里落地过一个是内部报表平台一个是日志分析系统。最大的体会是Spring Boot 和 Spark 的整合难点不在代码而在依赖管理和生命周期控制。代码本身很简单但版本冲突、内存泄漏、任务状态管理这些细节才是真正花时间的地方。另一个体会是不要试图用一套配置打天下。local 模式和集群模式的差异很大我现在的做法是配置文件分环境application-local.yml和application-cluster.yml分开启动时用--spring.profiles.active切换。这样本地调试和生产部署互不干扰。后续如果要扩展我会考虑这几个方向一是加任务队列和优先级用 Redis 或数据库存任务状态支持分布式部署二是加任务重试和告警失败后自动重试并通知三是把 Spark 任务的结果做成流式返回用 SSE 或 WebSocket 推送进度提升用户体验。最后分享一个小技巧在开发阶段把 Spark 的日志级别调到 WARN同时开启 Spark UI 的本地访问。local 模式下 Spark UI 默认在 4040 端口通过http://localhost:4040就能看到任务执行详情比看日志直观得多。这个习惯帮我省了大量排查时间。 
长河君 · 2026/10/17 16:29:02 · 阅读 1.2k Spring Boot整合Spark实战：HTTP接口调用与任务管理
1. 项目缘起与整体架构思路把 Spark 塞进 Spring Boot 里再用 HTTP 接口去触发任务这个需求我第一次碰到是在一个数据清洗与指标计算的后台系统里。当时的场景很典型业务侧有一堆 CSV 和日志文件需要做聚合统计算法同事用 Spark 写好了逻辑但运维和前端不想碰命令行希望点个按钮就能跑任务、看进度、拿结果。于是就有了这套“Spring Boot Spark HTTP 调用”的组合。先说清楚这个项目到底解决什么问题。传统 Spark 作业的提交方式是spark-submit脚本化、命令行化适合调度平台但对业务系统不友好。而 Spring Boot 天生适合做 Web 服务把 Spark 的SparkSession生命周期交给 Spring 容器管理对外暴露 REST 接口就能实现“请求即任务”的效果。这套方案适合谁适合有一定 Java 基础、想快速把离线计算能力嵌入到业务系统里的后端开发者也适合做数据中台、报表平台、批处理调度这类项目的同学。整体架构上我采用的是分层设计。最上层是 Controller负责接收 HTTP 请求、校验参数、返回任务 ID中间是 Service 层封装 Spark 任务的提交、状态查询和结果获取底层是 Spark 的SparkSession和具体的业务计算逻辑。这里有个关键决策点SparkSession是单例还是每次请求新建我的经验是在 local 模式下可以做成单例常驻因为 local 模式本身就是本地 JVM 内的线程池反复创建销毁开销大而在集群模式下更推荐每次任务独立提交避免资源长期占用和状态污染。为什么这么设计因为 Spark 的SparkContext一旦停止就无法重启单例模式下如果某个任务异常导致 context 关闭整个服务就废了。所以我在实际项目里做了一个折中local 模式用单例加任务队列集群模式用SparkLauncher或 REST 提交到集群Spring Boot 只做任务管理和状态轮询。这样既保证了本地调试的便利性又兼顾了生产环境的稳定性。还有一个容易被忽略的点是依赖冲突。Spring Boot 自带的日志、Jackson、Netty 等组件和 Spark 的依赖经常打架尤其是slf4j和log4j的版本问题。我的做法是在pom.xml里显式排除 Spark 自带的日志实现统一用 Spring Boot 的日志体系同时在打包时用maven-shade-plugin做重定位避免类加载冲突。这一步不做后面启动时报NoSuchMethodError会让你怀疑人生。2. 环境搭建与依赖配置的实操细节2.1 版本选型与兼容性对照版本选型是这个项目的第一道坎。Spring Boot 和 Spark 的版本不是随便配的Scala 版本、JDK 版本、Hadoop 版本之间有一张隐形的兼容表。我踩过的坑是Spring Boot 2.7 配 Spark 3.2 时JDK 必须用 8 或 11用 17 会遇到IllegalAccessError因为 Spark 内部大量使用了反射访问 JDK 内部类。下面这张表是我实测下来比较稳的组合Spring Boot 版本Spark 版本Scala 版本JDK 版本备注2.7.x3.2.x2.128 / 11最稳定推荐2.7.x3.3.x2.128 / 11需注意 Jackson 冲突3.0.x3.4.x2.1211 / 17需加 JVM 参数开放模块2.6.x3.1.x2.128老项目维护用选 Scala 2.12 而不是 2.13是因为大部分生产环境的 Hadoop 生态还是 2.12 为主2.13 的包相对少。JDK 方面如果你非要用 17记得在启动参数里加上--add-opens java.base/java.langALL-UNNAMED这类开放模块的配置否则 Spark 的序列化会直接报错。2.2 Maven 依赖的取舍与排除依赖配置我直接给一份可用的骨架重点看exclusions部分dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.2.4/version exclusions exclusion groupIdorg.slf4j/groupId artifactIdslf4j-log4j12/artifactId /exclusion exclusion groupIdlog4j/groupId artifactIdlog4j/artifactId /exclusion exclusion groupIdorg.codehaus.janino/groupId artifactIdjanino/artifactId /exclusion /exclusions /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.2.4/version /dependency为什么要排除janino因为 Spring Boot 的spring-boot-devtools或某些模板引擎也会引入 janino版本不一致时会导致 Spark SQL 的代码生成失败。排除后由 Spring Boot 统一管理版本问题就消失了。另外spark-sql是必须的因为实际业务里没人只用 RDDDataFrame 和 Dataset 才是主流。如果你还要用 Spark Streaming再加spark-streaming但注意流式任务和 Web 服务的生命周期管理会更复杂建议单独拆服务。2.3 配置文件的关键参数application.yml里我一般会放这些配置spark: app-name: spring-boot-spark-demo master: local[*] driver-memory: 2g executor-memory: 2g executor-cores: 2 sql-shuffle-partitions: 200 serializer: org.apache.spark.serializer.KryoSerializer kryo-registrator: com.example.MyKryoRegistratormaster在 local 模式下写local[*]表示用所有可用核心调试时可以用local[2]限制核心数方便观察资源占用。sql-shuffle-partitions默认是 200小数据量下这个值太大会产生大量小文件我一般会根据数据量调到 20 到 50 之间。序列化器强烈建议用 Kryo比 Java 原生序列化快好几倍但记得注册自定义类否则性能提升有限。提示driver-memory在 local 模式下其实就是 JVM 堆内存设置时不要超过物理内存的 70%否则容易触发 OOM Killer。3. SparkSession 的 Spring 化管理与 HTTP 接口设计3.1 SparkSession 单例 Bean 的正确写法把SparkSession交给 Spring 管理核心是写一个Configuration类Configuration public class SparkConfig { Value(${spark.app-name}) private String appName; Value(${spark.master}) private String master; Bean(destroyMethod stop) public SparkSession sparkSession() { return SparkSession.builder() .appName(appName) .master(master) .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) .config(spark.sql.shuffle.partitions, 50) .getOrCreate(); } }这里destroyMethod stop很关键Spring 容器关闭时会自动调用sparkSession.stop()避免 JVM 无法退出。我见过有人忘了加这个结果服务停了但 Spark 的线程还在跑进程一直挂着。但要注意SparkSession是重量级对象创建一次要几秒钟。如果你的服务是多实例部署每个实例都会创建一个 SparkSessionlocal 模式下没问题集群模式下会各自连集群资源消耗要提前评估。3.2 任务提交接口的设计HTTP 接口我设计成三个提交任务、查询状态、获取结果。提交接口用 POST返回一个任务 IDRestController RequestMapping(/api/spark) public class SparkJobController { Autowired private SparkJobService sparkJobService; PostMapping(/submit) public ResponseEntityString submit(RequestBody JobRequest request) { String jobId sparkJobService.submitJob(request); return ResponseEntity.ok(jobId); } GetMapping(/status/{jobId}) public ResponseEntityJobStatus status(PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getStatus(jobId)); } GetMapping(/result/{jobId}) public ResponseEntityListMapString, Object result(PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getResult(jobId)); } }为什么用异步提交而不是同步等待因为 Spark 任务可能跑几分钟甚至几小时HTTP 连接扛不住。我的做法是用ExecutorService或者CompletableFuture异步执行任务状态存在ConcurrentHashMap里前端轮询状态接口。任务 ID 用 UUID简单可靠。JobRequest里一般包含数据源路径、输出路径、计算类型、过滤条件等参数。这里有个安全考虑绝对不要让用户直接传 SQL 或代码片段否则就是注入漏洞。我的做法是预定义几种计算模板用户只传参数模板在服务端映射成具体的 DataFrame 操作。3.3 任务状态管理与结果缓存状态管理我用一个简单的枚举PENDING、RUNNING、SUCCESS、FAILED。每个任务对应一个JobContext对象包含状态、开始时间、结束时间、错误信息、结果集。结果集我一般限制条数比如最多返回 1000 行避免内存爆掉。public class JobContext { private String jobId; private volatile JobStatus status; private long startTime; private long endTime; private String errorMsg; private ListMapString, Object result; }结果缓存用ConcurrentHashMap就够了但如果任务量大建议加过期清理策略比如任务完成 1 小时后自动移除。我试过用 Caffeine 做本地缓存效果不错配置expireAfterWrite(1, TimeUnit.HOURS)即可。注意结果集不要直接存 DataFrameDataFrame 持有 Spark 的 lineage 和资源引用存多了会导致内存泄漏。一定要collectAsList()转成普通 Java 集合再存。4. 集群模式与 local 模式的切换与优化4.1 两种模式的本质区别local 模式和集群模式的区别不只是master参数不同。local 模式下driver 和 executor 在同一个 JVM 里共享内存和 CPU调试方便但资源隔离差。集群模式下driver 在 Spring Boot 服务里executor 在集群的 worker 节点上通过网络通信资源隔离好但延迟高。我一般这样选开发和测试用 local生产环境用集群。但有些场景下数据量不大、计算逻辑简单local 模式反而更快因为省去了网络传输和任务调度开销。判断标准是数据量超过单机内存的 50%或者计算需要多节点并行就上集群。集群模式下Spring Boot 服务本身不跑计算只做任务提交和状态管理。提交方式有两种一种是用SparkLauncher在本地启动一个提交进程另一种是直接调集群的 REST 接口。SparkLauncher的好处是能拿到任务句柄方便监控坏处是依赖本地有 Spark 客户端环境。SparkAppHandle handle new SparkLauncher() .setAppResource(/path/to/job.jar) .setMainClass(com.example.SparkJob) .setMaster(yarn) .setDeployMode(cluster) .addAppArgs(jobId, inputPath, outputPath) .startApplication();这段代码在 local 模式下也能跑但setMaster(yarn)需要 Hadoop 配置。如果你的集群是 Standalone 模式改成spark://host:7077即可。4.2 资源配置的计算过程资源配置是优化的核心。假设你的集群有 3 个节点每个节点 16 核 64G 内存那么总资源是 48 核 192G。一般建议 executor 数量不超过总核数的 80%留一部分给系统和其他服务。我常用的配置公式是executor 数量 总核数 / 每个 executor 的核数每个 executor 的核数建议 2 到 5太多会导致 HDFS IO 竞争executor 内存 节点内存 / 每节点 executor 数量 * 0.8比如 3 节点 16 核 64G我配置 6 个 executor每个 4 核 16G这样用了 24 核 96G留了一半资源给系统。driver-memory一般 2G 到 4G 足够除非你要collect大量数据到 driver。spark: executor: instances: 6 cores: 4 memory: 16g driver: memory: 4g这些参数在 local 模式下大部分无效因为 local 模式只有一个 JVM。所以我在代码里做了判断如果是 local 模式忽略 executor 相关配置只设置 driver 内存。4.3 性能优化的几个关键点第一个优化点是分区数。spark.sql.shuffle.partitions默认 200小数据量下会产生大量小任务调度开销比计算还大。我的经验是数据量小于 1G 时设 10 到 201G 到 10G 设 50 到 10010G 以上再考虑 200 以上。第二个优化点是缓存策略。如果同一个 DataFrame 被多次使用一定要cache()或persist()。但不要滥用缓存会占内存用完后unpersist()释放。我见过有人把所有 DataFrame 都 cache结果 executor OOM。第三个优化点是广播变量。小表 join 大表时用broadcast()把小表广播出去避免 shuffle。这个优化效果非常明显我实测过一个 10G 大表 join 100M 小表的场景广播后耗时从 8 分钟降到 1 分钟。DatasetRow result largeTable.join( broadcast(smallTable), joinKey );第四个优化点是序列化。Kryo 序列化比 Java 原生快 2 到 5 倍但需要注册类。我一般会写一个KryoRegistrator把所有自定义的 POJO 都注册进去。提示优化不是一次性的要结合 Spark UI 看每个 stage 的耗时和 shuffle 数据量针对性调整。盲目调参不如先看监控。5. 常见问题排查与避坑经验5.1 启动阶段的典型报错问题一java.lang.NoClassDefFoundError: org/slf4j/Logger这是日志冲突的典型表现。原因是 Spark 自带的 slf4j 版本和 Spring Boot 的不一致。解决办法是在pom.xml里排除 Spark 的slf4j-log4j12统一用 Spring Boot 的spring-boot-starter-logging。问题二java.lang.IllegalAccessError: class org.apache.spark... cannot access class sun.nio.ch...这是 JDK 17 的模块化限制导致的。解决办法是加 JVM 参数--add-opens java.base/sun.nio.chALL-UNNAMED --add-opens java.base/java.langALL-UNNAMED --add-opens java.base/java.utilALL-UNNAMED问题三org.apache.spark.SparkException: A master URL must be set说明SparkSession创建时没有设置 master。检查配置文件是否被正确加载或者代码里是否硬编码了master。5.2 运行阶段的状态异常问题四任务一直卡在 RUNNING 状态可能原因有三个一是数据倾斜某个 key 的数据量特别大二是资源不足executor 申请不到三是死锁比如在 Spark 任务里调用了 Web 服务的同步接口。排查方法是看 Spark UI 的 stage 详情找到卡住的 task。问题五OutOfMemoryError: Java heap spacedriver OOM 一般是collect()数据量太大改成take()或写文件。executor OOM 一般是分区数据不均匀或缓存太多调整分区数或减少缓存。问题六HTTP 请求超时Spark 任务执行时间超过网关的超时限制。解决办法是改成异步提交前端轮询状态。如果必须同步把网关超时调到 10 分钟以上但不推荐。5.3 常见问题速查表问题现象可能原因排查方法解决方案启动报 NoClassDefFoundError依赖冲突mvn dependency:tree排除冲突包JDK 17 反射报错模块化限制看报错类名加 add-opens 参数任务卡住不动数据倾斜/资源不足Spark UI 看 stage加盐/调资源driver OOMcollect 数据过多看 driver 日志改 take 或写文件executor OOM缓存过多/分区不均看 executor 日志减缓存/调分区结果返回慢结果集太大看接口耗时限制返回条数服务无法退出SparkSession 未关闭jstack 看线程加 destroyMethod5.4 独家避坑技巧第一个技巧在 Spring Boot 启动类上加EnableAsync把 Spark 任务放到独立线程池。不要用默认的SimpleAsyncTaskExecutor那个每次请求都新建线程任务多了会爆。自定义一个ThreadPoolTaskExecutor核心线程数设 5 到 10队列容量设 100。第二个技巧给 Spark 任务加超时控制。用Future.get(timeout, TimeUnit)包一层超时后调用future.cancel(true)并记录日志。但注意 Spark 任务不一定能立即中断所以还要配合sparkSession.stop()或SparkContext.cancelJobGroup()。第三个技巧日志分级。Spark 的日志非常啰嗦默认 INFO 级别会刷屏。在logback-spring.xml里把org.apache.spark设为 WARN只保留关键信息。调试时再临时开 DEBUG。logger nameorg.apache.spark levelWARN/ logger nameorg.apache.hadoop levelWARN/第四个技巧本地模式下的临时目录。Spark 会在/tmp下创建大量临时文件时间长了会占满磁盘。在spark-defaults.conf或代码里设置spark.local.dir到一个大容量目录并定期清理。.config(spark.local.dir, /data/spark-tmp)第五个技巧任务幂等性。同一个任务重复提交时要保证结果一致。我的做法是输出路径带任务 ID或者写入前先删除旧数据。否则重跑任务会报“路径已存在”的错误。6. 完整实操流程与代码骨架6.1 从零搭建的步骤清单第一步创建 Spring Boot 项目选 Maven、JDK 8 或 11引入spring-boot-starter-web和spark-core、spark-sql。第二步配置pom.xml排除日志冲突加maven-shade-plugin做打包。第三步写SparkConfig配置类创建SparkSessionBean。第四步写SparkJobService实现任务提交、状态管理、结果缓存。第五步写SparkJobController暴露三个 HTTP 接口。第六步写具体的 Spark 计算逻辑比如 CSV 读取、聚合、写回。第七步本地用local[*]测试通过后改成集群模式部署。6.2 核心 Service 的代码实现Service public class SparkJobService { Autowired private SparkSession sparkSession; private final MapString, JobContext jobMap new ConcurrentHashMap(); private final ExecutorService executor Executors.newFixedThreadPool(5); public String submitJob(JobRequest request) { String jobId UUID.randomUUID().toString(); JobContext context new JobContext(jobId); jobMap.put(jobId, context); executor.submit(() - { context.setStatus(JobStatus.RUNNING); context.setStartTime(System.currentTimeMillis()); try { ListMapString, Object result executeJob(request); context.setResult(result); context.setStatus(JobStatus.SUCCESS); } catch (Exception e) { context.setErrorMsg(e.getMessage()); context.setStatus(JobStatus.FAILED); } finally { context.setEndTime(System.currentTimeMillis()); } }); return jobId; } private ListMapString, Object executeJob(JobRequest request) { DatasetRow df sparkSession.read() .option(header, true) .csv(request.getInputPath()); DatasetRow result df.groupBy(category) .agg(functions.sum(amount).as(total)); return result.limit(1000).collectAsList().stream() .map(row - row.getValuesMap(row.schema().fieldNames())) .collect(Collectors.toList()); } }这段代码是骨架实际业务里executeJob会根据request.getJobType()走不同的分支。我一般用策略模式每种任务类型对应一个JobHandler实现类Service 里根据类型路由。6.3 集群模式下的提交改造集群模式下executeJob不再直接跑计算而是提交到集群。改造后的逻辑private ListMapString, Object executeJob(JobRequest request) throws Exception { SparkAppHandle handle new SparkLauncher() .setAppResource(request.getJarPath()) .setMainClass(request.getMainClass()) .setMaster(request.getMaster()) .setDeployMode(cluster) .addAppArgs(request.getInputPath(), request.getOutputPath()) .startApplication(); while (!handle.getState().isFinal()) { Thread.sleep(2000); } if (handle.getState() SparkAppHandle.State.FINISHED) { return readResultFromOutput(request.getOutputPath()); } else { throw new RuntimeException(Spark job failed: handle.getState()); } }这里的结果读取是从输出路径读文件而不是从 Spark 直接返回。因为集群模式下 driver 和 executor 分离结果一般写 HDFS 或对象存储再由 Spring Boot 服务读取。6.4 参数调优的实测记录我拿一个 5G 的 CSV 文件做过对比测试任务是按类别聚合金额。测试环境是 3 节点每节点 8 核 32G。配置项默认值调优值耗时对比shuffle partitions200504分20秒 → 2分50秒序列化器JavaKryo2分50秒 → 2分10秒executor 内存1g4g频繁 GC → 稳定广播 join未用使用小表 join 快 3 倍调优后总耗时从 6 分钟降到 2 分半效果还是很明显的。但要注意调优不是越多越好比如 executor 内存设太大会导致 GC 时间变长反而变慢。我的经验是先看 Spark UI 找瓶颈再针对性调参每次只改一个参数记录对比结果。7. 个人实操体会与后续扩展方向这套方案我在两个项目里落地过一个是内部报表平台一个是日志分析系统。最大的体会是Spring Boot 和 Spark 的整合难点不在代码而在依赖管理和生命周期控制。代码本身很简单但版本冲突、内存泄漏、任务状态管理这些细节才是真正花时间的地方。另一个体会是不要试图用一套配置打天下。local 模式和集群模式的差异很大我现在的做法是配置文件分环境application-local.yml和application-cluster.yml分开启动时用--spring.profiles.active切换。这样本地调试和生产部署互不干扰。后续如果要扩展我会考虑这几个方向一是加任务队列和优先级用 Redis 或数据库存任务状态支持分布式部署二是加任务重试和告警失败后自动重试并通知三是把 Spark 任务的结果做成流式返回用 SSE 或 WebSocket 推送进度提升用户体验。最后分享一个小技巧在开发阶段把 Spark 的日志级别调到 WARN同时开启 Spark UI 的本地访问。local 模式下 Spark UI 默认在 4040 端口通过http://localhost:4040就能看到任务执行详情比看日志直观得多。这个习惯帮我省了大量排查时间。