1. 为什么要在 Spring Boot 里整合 Spark把 Spark 塞进 Spring Boot 这件事最早我是拒绝的。原因很简单Spark 自己就是个完整的计算框架有 Driver、有 Executor、有集群管理Spring Boot 是个 Web 应用框架两者定位差得挺远。但实际项目里业务系统往往是 Spring Boot 那一套数据侧又需要跑 Spark 做批量计算或者特征处理如果让业务系统直接去调spark-submit脚本运维和参数管理会变得非常难受。所以整合的核心诉求其实就三个用 HTTP 接口触发 Spark 任务、任务参数和 Spark 配置能动态下发、local 模式和集群模式能一套代码切换。这三点做到了业务侧就只需要发一个 POST 请求剩下的 Spark 生命周期全部由后端托管。这套方案适合谁我总结下来是这几类场景数据平台需要给业务方提供计算即服务的接口中小团队没有独立的大数据调度系统想用 Spring Boot 当轻量级调度入口还有做算法服务的同学想把特征工程、模型训练这类 Spark 任务包装成 REST 接口给前端或上游调用。需要提前说清楚一个认知Spring Boot 整合 Spark 不是把 Spark 变成 Spring 的 Bean 那么简单。Spark 的SparkSession是有状态、重量级的对象它的生命周期管理和 Spring 的 singleton Bean 模型并不天然契合。这一点如果没想明白后面会遇到一堆内存泄漏和任务串扰的问题。下面我会把整个设计思路、配置细节、实操步骤和踩过的坑都摊开讲。2. 整体架构设计与模式选型思路2.1 两种运行模式的本质区别local 模式和集群模式在代码层面最大的差异不是 API而是SparkSession的构建方式和依赖的提交机制。local 模式下Driver 和 Executor 都在同一个 JVM 进程里master设为local[*]就行调试极其方便。集群模式下Driver 可能在本地client 模式也可能在集群里cluster 模式Executor 分布在多台机器上任务提交需要通过集群管理器。我一般这样划分使用场景模式适用阶段优点缺点local开发调试、小数据量验证启动快、日志全、断点方便无法压测、资源受限client 集群生产常规任务Driver 日志本地可见、便于排查Driver 占用提交机资源cluster 集群长时任务、生产稳定Driver 高可用、不占提交机日志需去集群捞、调试麻烦选型逻辑很直接开发期一律 local生产期优先 client 模式。cluster 模式虽然更正统但日志排查成本高除非任务跑几个小时以上、Driver 挂了需要自动重启否则 client 模式性价比更高。2.2 为什么用 HTTP 接口而不是命令行提交命令行spark-submit的问题在于参数写死在脚本里、任务状态无法实时回传、多租户场景下没法做权限隔离。用 HTTP 接口封装之后任务参数可以走 JSON 下发任务 ID 可以返回给调用方轮询状态Spark 配置也能按任务粒度覆盖。这里有个关键设计决策SparkSession 是复用还是每次新建。我的经验是分两种情况——如果是交互式查询、低延迟场景可以维护一个长生命周期的 Session 池如果是批处理任务每次新建 Session、任务结束就stop()避免状态污染。后者更稳妥代价是启动开销local 模式几秒集群模式十几秒到几十秒。2.3 整体分层结构我把整个模块拆成四层职责清晰接口层RestController接收 HTTP 请求做参数校验和任务 ID 生成任务管理层维护任务状态提交中/运行中/成功/失败提供查询接口Spark 执行层负责构建SparkSession、加载配置、执行具体业务逻辑配置层管理 Spark 默认配置和任务级覆盖配置这样拆的好处是Spark 相关的代码全部收敛在执行层业务逻辑通过接口注入测试时可以 mock 掉执行层不用真的起 Spark。3. 核心依赖与 SparkSession 构建细节3.1 依赖引入的版本对齐问题Spring Boot 和 Spark 的版本兼容是个大坑。Spark 3.x 依赖 Scala 2.12 或 2.13而 Spring Boot 的某些 starter 会间接引入不同版本的 Jackson、Guava导致运行时NoSuchMethodError。我的做法是显式排除冲突依赖锁定关键库版本。dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.5.0/version exclusions exclusion groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /exclusion exclusion groupIdcom.google.guava/groupId artifactIdguava/artifactId /exclusion /exclusions /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.5.0/version /dependency注意spark-core和spark-sql的 Scala 版本后缀必须一致_2.12和_2.13混用会直接报类找不到。打包时如果用spring-boot-maven-plugin记得把 Spark 相关依赖设为provided或用shade插件重定位否则 fat jar 会大到离谱几百 MB。3.2 SparkSession 构建的两种写法local 模式的构建很直接SparkSession session SparkSession.builder() .appName(local-job- taskId) .master(local[*]) .config(spark.sql.shuffle.partitions, 4) .config(spark.driver.memory, 2g) .getOrCreate();集群模式则不能硬编码master而是通过spark-submit的--master参数传入代码里要么不设要么从配置读取SparkSession session SparkSession.builder() .appName(cluster-job- taskId) .config(conf) // 从外部配置加载 .getOrCreate();这里有个细节getOrCreate()在同一个 JVM 里会复用已有的 Session。如果你在 Web 应用里多次调用第二次拿到的还是第一次的 Session配置不会生效。所以批处理场景我强烈建议每次任务结束后调用session.stop()或者用newSession()创建独立会话。3.3 配置参数的动态下发任务级配置覆盖是这套方案的核心价值。我设计了一个SparkJobConfig对象包含master、executorMemory、executorCores、numExecutors、extraConf等字段通过 HTTP body 传入。执行层把它转换成SparkConfSparkConf conf new SparkConf(); conf.set(spark.executor.memory, config.getExecutorMemory()); conf.set(spark.executor.cores, String.valueOf(config.getExecutorCores())); conf.set(spark.executor.instances, String.valueOf(config.getNumExecutors())); config.getExtraConf().forEach(conf::set);提示spark.executor.instances在 local 模式下无效设了也不报错但会被忽略。集群模式下如果用的是 YARN 动态资源分配这个参数也要关掉动态分配才生效否则以spark.dynamicAllocation为准。4. HTTP 接口设计与任务生命周期管理4.1 接口定义与参数结构我一般暴露三个接口提交任务、查询状态、取消任务。提交接口的请求体长这样{ jobType: wordCount, inputPath: /data/input, outputPath: /data/output, sparkConfig: { master: local[*], executorMemory: 2g, executorCores: 2, numExecutors: 4, extraConf: { spark.sql.shuffle.partitions: 200 } } }jobType用来路由到具体的业务逻辑我用一个MapString, SparkJobHandler注册所有 handler新增任务类型只需要加一个实现类符合开闭原则。4.2 异步执行与任务状态维护HTTP 请求不能同步等 Spark 任务跑完否则连接超时是必然的。所以提交接口只负责把任务丢进线程池立即返回任务 IDString taskId UUID.randomUUID().toString(); taskStatusMap.put(taskId, TaskStatus.SUBMITTED); executorService.submit(() - { try { taskStatusMap.put(taskId, TaskStatus.RUNNING); handler.execute(taskId, request); taskStatusMap.put(taskId, TaskStatus.SUCCESS); } catch (Exception e) { taskStatusMap.put(taskId, TaskStatus.FAILED); log.error(task {} failed, taskId, e); } }); return taskId;线程池的配置有讲究不能用Executors.newFixedThreadPool无界队列任务堆积会 OOM。我一般用ThreadPoolExecutor显式指定队列容量和拒绝策略容量按机器内存和单任务资源占用估算。4.3 任务状态存储的选择内存ConcurrentHashMap最简单但应用重启就丢了。生产环境我建议落库或者用 Redis字段至少包含任务 ID、状态、提交时间、开始时间、结束时间、错误信息、Spark Application ID。Spark Application ID 特别重要集群模式下排查问题全靠它去集群 UI 里定位。实操心得任务状态更新要加时间戳并且查询接口返回时计算一下运行时长。有次线上任务卡住就是因为没有时长监控等业务方反馈才发现已经挂了两个小时。5. Spark 配置优化与参数调优实战5.1 资源配置的计算逻辑Executor 资源配置不是拍脑袋定的有个基本公式。假设集群有 N 台机器每台 M 核、G 内存预留 1 核 1G 给系统每台可用的 executor 数 (M - 1) / (cores_per_executor 1)每个 executor 内存 (G - 1) / executor_per_node举个例子3 台 16 核 64G 的机器想每个 executor 用 4 核每台 executor 数 (16 - 1) / (4 1) 3总 executor 数 3 × 3 9每 executor 内存 (64 - 1) / 3 ≈ 21G取 20G 留余量这个计算过程我一般写在配置校验里如果用户传的参数明显不合理比如 executor 内存超过单机内存直接拒绝并提示。5.2 常见调优参数速查参数作用经验值spark.sql.shuffle.partitionsshuffle 分区数数据量 GB 级设 200-500spark.default.parallelismRDD 默认并行度executor 核数 × 2~3spark.executor.memoryOverhead堆外内存executor 内存的 10%spark.serializer序列化器KryoSerializerspark.sql.adaptive.enabled自适应执行truespark.sql.shuffle.partitions默认 200小数据量下会产生大量小任务调度开销比计算还大。我一般按数据量 / 128MB估算再结合spark.sql.adaptive让 Spark 自己合并小分区。5.3 序列化的坑默认 Java 序列化又慢又占空间换成 Kryo 能省 30% 以上内存。但 Kryo 需要注册自定义类否则性能提升有限conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer); conf.set(spark.kryo.registrationRequired, false); conf.registerKryoClasses(new Class[]{MyBean.class});注意registrationRequired设为 true 时未注册的类会直接报错调试期建议设 false生产期再收紧。6. 常见问题排查与避坑经验6.1 典型问题速查表现象可能原因排查方向任务提交后无响应线程池满 / Session 构建卡住看线程栈、检查 master 地址NoSuchMethodError依赖版本冲突mvn dependency:tree找冲突OOM in Driver结果集 collect 到本地改用 write 落盘别 collect任务一直 RUNNING数据倾斜 / 资源不足看 Spark UI 的 stage 分布local 模式正常集群报错路径/依赖在集群节点不存在检查 HDFS 路径和 jar 分发6.2 数据倾斜的识别与处理数据倾斜最典型的症状是大部分 task 秒完少数几个 task 跑几十分钟。识别方法是看 Spark UI 里 stage 的 task 耗时分布。处理手段我常用三种加盐打散 key、repartition重分区、广播小表做 map join。加盐适合聚合场景广播适合 join 场景重分区是通用兜底但治标不治本。6.3 我踩过的几个坑第一个坑是在 Web 容器里没关 Session。Tomcat 的线程池是复用的每次请求都getOrCreate拿到同一个 Session配置互相覆盖任务结果串了。后来改成每个任务独立 Session 并显式 stop 才解决。第二个坑是日志配置冲突。Spark 自带 log4j 配置和 Spring Boot 的 logback 打架导致 Spark 日志全丢。解决办法是在application.yml里排除 Spark 的日志依赖或者用spark-submit时指定--files传自定义 log4j 配置。第三个坑是集群模式下 jar 包分发。本地跑得好好的提交到集群就报类找不到原因是业务类没打进提交的 jar 里。用--jars显式指定或者打成 fat jar 提交。7. 一套可复用的落地建议如果你准备动手做这套整合我的建议是先把 local 模式跑通用一个最简单的 word count 验证接口到执行的完整链路再逐步加集群配置。配置管理上把 Spark 默认配置和任务级配置分开默认配置放application.yml任务级走请求参数覆盖这样既保证一致性又保留灵活性。资源隔离方面如果多个任务并发跑一定要限制线程池大小和单任务资源上限否则一个失控的任务能把整个应用拖垮。监控上至少记录任务提交数、成功率、平均耗时三个指标接入现有的监控体系。最后说个容易被忽略的点Spark 任务的幂等性。HTTP 接口可能被重试同一个任务提交两次会怎样我的做法是让调用方传一个业务唯一键服务端做去重重复提交直接返回已有任务 ID。这个设计在对接上游系统时能省掉大量扯皮。 
长河君 · 2026/10/17 16:29:02 · 阅读 1.2k Spring Boot 整合 Spark：HTTP 接口触发与集群模式动态配置实战
1. 为什么要在 Spring Boot 里整合 Spark把 Spark 塞进 Spring Boot 这件事最早我是拒绝的。原因很简单Spark 自己就是个完整的计算框架有 Driver、有 Executor、有集群管理Spring Boot 是个 Web 应用框架两者定位差得挺远。但实际项目里业务系统往往是 Spring Boot 那一套数据侧又需要跑 Spark 做批量计算或者特征处理如果让业务系统直接去调spark-submit脚本运维和参数管理会变得非常难受。所以整合的核心诉求其实就三个用 HTTP 接口触发 Spark 任务、任务参数和 Spark 配置能动态下发、local 模式和集群模式能一套代码切换。这三点做到了业务侧就只需要发一个 POST 请求剩下的 Spark 生命周期全部由后端托管。这套方案适合谁我总结下来是这几类场景数据平台需要给业务方提供计算即服务的接口中小团队没有独立的大数据调度系统想用 Spring Boot 当轻量级调度入口还有做算法服务的同学想把特征工程、模型训练这类 Spark 任务包装成 REST 接口给前端或上游调用。需要提前说清楚一个认知Spring Boot 整合 Spark 不是把 Spark 变成 Spring 的 Bean 那么简单。Spark 的SparkSession是有状态、重量级的对象它的生命周期管理和 Spring 的 singleton Bean 模型并不天然契合。这一点如果没想明白后面会遇到一堆内存泄漏和任务串扰的问题。下面我会把整个设计思路、配置细节、实操步骤和踩过的坑都摊开讲。2. 整体架构设计与模式选型思路2.1 两种运行模式的本质区别local 模式和集群模式在代码层面最大的差异不是 API而是SparkSession的构建方式和依赖的提交机制。local 模式下Driver 和 Executor 都在同一个 JVM 进程里master设为local[*]就行调试极其方便。集群模式下Driver 可能在本地client 模式也可能在集群里cluster 模式Executor 分布在多台机器上任务提交需要通过集群管理器。我一般这样划分使用场景模式适用阶段优点缺点local开发调试、小数据量验证启动快、日志全、断点方便无法压测、资源受限client 集群生产常规任务Driver 日志本地可见、便于排查Driver 占用提交机资源cluster 集群长时任务、生产稳定Driver 高可用、不占提交机日志需去集群捞、调试麻烦选型逻辑很直接开发期一律 local生产期优先 client 模式。cluster 模式虽然更正统但日志排查成本高除非任务跑几个小时以上、Driver 挂了需要自动重启否则 client 模式性价比更高。2.2 为什么用 HTTP 接口而不是命令行提交命令行spark-submit的问题在于参数写死在脚本里、任务状态无法实时回传、多租户场景下没法做权限隔离。用 HTTP 接口封装之后任务参数可以走 JSON 下发任务 ID 可以返回给调用方轮询状态Spark 配置也能按任务粒度覆盖。这里有个关键设计决策SparkSession 是复用还是每次新建。我的经验是分两种情况——如果是交互式查询、低延迟场景可以维护一个长生命周期的 Session 池如果是批处理任务每次新建 Session、任务结束就stop()避免状态污染。后者更稳妥代价是启动开销local 模式几秒集群模式十几秒到几十秒。2.3 整体分层结构我把整个模块拆成四层职责清晰接口层RestController接收 HTTP 请求做参数校验和任务 ID 生成任务管理层维护任务状态提交中/运行中/成功/失败提供查询接口Spark 执行层负责构建SparkSession、加载配置、执行具体业务逻辑配置层管理 Spark 默认配置和任务级覆盖配置这样拆的好处是Spark 相关的代码全部收敛在执行层业务逻辑通过接口注入测试时可以 mock 掉执行层不用真的起 Spark。3. 核心依赖与 SparkSession 构建细节3.1 依赖引入的版本对齐问题Spring Boot 和 Spark 的版本兼容是个大坑。Spark 3.x 依赖 Scala 2.12 或 2.13而 Spring Boot 的某些 starter 会间接引入不同版本的 Jackson、Guava导致运行时NoSuchMethodError。我的做法是显式排除冲突依赖锁定关键库版本。dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.5.0/version exclusions exclusion groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId /exclusion exclusion groupIdcom.google.guava/groupId artifactIdguava/artifactId /exclusion /exclusions /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-sql_2.12/artifactId version3.5.0/version /dependency注意spark-core和spark-sql的 Scala 版本后缀必须一致_2.12和_2.13混用会直接报类找不到。打包时如果用spring-boot-maven-plugin记得把 Spark 相关依赖设为provided或用shade插件重定位否则 fat jar 会大到离谱几百 MB。3.2 SparkSession 构建的两种写法local 模式的构建很直接SparkSession session SparkSession.builder() .appName(local-job- taskId) .master(local[*]) .config(spark.sql.shuffle.partitions, 4) .config(spark.driver.memory, 2g) .getOrCreate();集群模式则不能硬编码master而是通过spark-submit的--master参数传入代码里要么不设要么从配置读取SparkSession session SparkSession.builder() .appName(cluster-job- taskId) .config(conf) // 从外部配置加载 .getOrCreate();这里有个细节getOrCreate()在同一个 JVM 里会复用已有的 Session。如果你在 Web 应用里多次调用第二次拿到的还是第一次的 Session配置不会生效。所以批处理场景我强烈建议每次任务结束后调用session.stop()或者用newSession()创建独立会话。3.3 配置参数的动态下发任务级配置覆盖是这套方案的核心价值。我设计了一个SparkJobConfig对象包含master、executorMemory、executorCores、numExecutors、extraConf等字段通过 HTTP body 传入。执行层把它转换成SparkConfSparkConf conf new SparkConf(); conf.set(spark.executor.memory, config.getExecutorMemory()); conf.set(spark.executor.cores, String.valueOf(config.getExecutorCores())); conf.set(spark.executor.instances, String.valueOf(config.getNumExecutors())); config.getExtraConf().forEach(conf::set);提示spark.executor.instances在 local 模式下无效设了也不报错但会被忽略。集群模式下如果用的是 YARN 动态资源分配这个参数也要关掉动态分配才生效否则以spark.dynamicAllocation为准。4. HTTP 接口设计与任务生命周期管理4.1 接口定义与参数结构我一般暴露三个接口提交任务、查询状态、取消任务。提交接口的请求体长这样{ jobType: wordCount, inputPath: /data/input, outputPath: /data/output, sparkConfig: { master: local[*], executorMemory: 2g, executorCores: 2, numExecutors: 4, extraConf: { spark.sql.shuffle.partitions: 200 } } }jobType用来路由到具体的业务逻辑我用一个MapString, SparkJobHandler注册所有 handler新增任务类型只需要加一个实现类符合开闭原则。4.2 异步执行与任务状态维护HTTP 请求不能同步等 Spark 任务跑完否则连接超时是必然的。所以提交接口只负责把任务丢进线程池立即返回任务 IDString taskId UUID.randomUUID().toString(); taskStatusMap.put(taskId, TaskStatus.SUBMITTED); executorService.submit(() - { try { taskStatusMap.put(taskId, TaskStatus.RUNNING); handler.execute(taskId, request); taskStatusMap.put(taskId, TaskStatus.SUCCESS); } catch (Exception e) { taskStatusMap.put(taskId, TaskStatus.FAILED); log.error(task {} failed, taskId, e); } }); return taskId;线程池的配置有讲究不能用Executors.newFixedThreadPool无界队列任务堆积会 OOM。我一般用ThreadPoolExecutor显式指定队列容量和拒绝策略容量按机器内存和单任务资源占用估算。4.3 任务状态存储的选择内存ConcurrentHashMap最简单但应用重启就丢了。生产环境我建议落库或者用 Redis字段至少包含任务 ID、状态、提交时间、开始时间、结束时间、错误信息、Spark Application ID。Spark Application ID 特别重要集群模式下排查问题全靠它去集群 UI 里定位。实操心得任务状态更新要加时间戳并且查询接口返回时计算一下运行时长。有次线上任务卡住就是因为没有时长监控等业务方反馈才发现已经挂了两个小时。5. Spark 配置优化与参数调优实战5.1 资源配置的计算逻辑Executor 资源配置不是拍脑袋定的有个基本公式。假设集群有 N 台机器每台 M 核、G 内存预留 1 核 1G 给系统每台可用的 executor 数 (M - 1) / (cores_per_executor 1)每个 executor 内存 (G - 1) / executor_per_node举个例子3 台 16 核 64G 的机器想每个 executor 用 4 核每台 executor 数 (16 - 1) / (4 1) 3总 executor 数 3 × 3 9每 executor 内存 (64 - 1) / 3 ≈ 21G取 20G 留余量这个计算过程我一般写在配置校验里如果用户传的参数明显不合理比如 executor 内存超过单机内存直接拒绝并提示。5.2 常见调优参数速查参数作用经验值spark.sql.shuffle.partitionsshuffle 分区数数据量 GB 级设 200-500spark.default.parallelismRDD 默认并行度executor 核数 × 2~3spark.executor.memoryOverhead堆外内存executor 内存的 10%spark.serializer序列化器KryoSerializerspark.sql.adaptive.enabled自适应执行truespark.sql.shuffle.partitions默认 200小数据量下会产生大量小任务调度开销比计算还大。我一般按数据量 / 128MB估算再结合spark.sql.adaptive让 Spark 自己合并小分区。5.3 序列化的坑默认 Java 序列化又慢又占空间换成 Kryo 能省 30% 以上内存。但 Kryo 需要注册自定义类否则性能提升有限conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer); conf.set(spark.kryo.registrationRequired, false); conf.registerKryoClasses(new Class[]{MyBean.class});注意registrationRequired设为 true 时未注册的类会直接报错调试期建议设 false生产期再收紧。6. 常见问题排查与避坑经验6.1 典型问题速查表现象可能原因排查方向任务提交后无响应线程池满 / Session 构建卡住看线程栈、检查 master 地址NoSuchMethodError依赖版本冲突mvn dependency:tree找冲突OOM in Driver结果集 collect 到本地改用 write 落盘别 collect任务一直 RUNNING数据倾斜 / 资源不足看 Spark UI 的 stage 分布local 模式正常集群报错路径/依赖在集群节点不存在检查 HDFS 路径和 jar 分发6.2 数据倾斜的识别与处理数据倾斜最典型的症状是大部分 task 秒完少数几个 task 跑几十分钟。识别方法是看 Spark UI 里 stage 的 task 耗时分布。处理手段我常用三种加盐打散 key、repartition重分区、广播小表做 map join。加盐适合聚合场景广播适合 join 场景重分区是通用兜底但治标不治本。6.3 我踩过的几个坑第一个坑是在 Web 容器里没关 Session。Tomcat 的线程池是复用的每次请求都getOrCreate拿到同一个 Session配置互相覆盖任务结果串了。后来改成每个任务独立 Session 并显式 stop 才解决。第二个坑是日志配置冲突。Spark 自带 log4j 配置和 Spring Boot 的 logback 打架导致 Spark 日志全丢。解决办法是在application.yml里排除 Spark 的日志依赖或者用spark-submit时指定--files传自定义 log4j 配置。第三个坑是集群模式下 jar 包分发。本地跑得好好的提交到集群就报类找不到原因是业务类没打进提交的 jar 里。用--jars显式指定或者打成 fat jar 提交。7. 一套可复用的落地建议如果你准备动手做这套整合我的建议是先把 local 模式跑通用一个最简单的 word count 验证接口到执行的完整链路再逐步加集群配置。配置管理上把 Spark 默认配置和任务级配置分开默认配置放application.yml任务级走请求参数覆盖这样既保证一致性又保留灵活性。资源隔离方面如果多个任务并发跑一定要限制线程池大小和单任务资源上限否则一个失控的任务能把整个应用拖垮。监控上至少记录任务提交数、成功率、平均耗时三个指标接入现有的监控体系。最后说个容易被忽略的点Spark 任务的幂等性。HTTP 接口可能被重试同一个任务提交两次会怎样我的做法是让调用方传一个业务唯一键服务端做去重重复提交直接返回已有任务 ID。这个设计在对接上游系统时能省掉大量扯皮。