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> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>3.5.0</version> <exclusions> <exclusion> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </exclusion> <exclusion> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.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 传入。执行层把它转换成SparkConf：
SparkConf 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用来路由到具体的业务逻辑，我用一个Map<String, SparkJobHandler>注册所有 handler，新增任务类型只需要加一个实现类，符合开闭原则。
4.2 异步执行与任务状态维护
HTTP 请求不能同步等 Spark 任务跑完，否则连接超时是必然的。所以提交接口只负责把任务丢进线程池，立即返回任务 ID：
String 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.partitions | shuffle 分区数 | 数据量 GB 级设 200-500 |
| spark.default.parallelism | RDD 默认并行度 | executor 核数 × 2~3 |
| spark.executor.memoryOverhead | 堆外内存 | executor 内存的 10% |
| spark.serializer | 序列化器 | KryoSerializer |
| spark.sql.adaptive.enabled | 自适应执行 | true |
spark.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。这个设计在对接上游系统时能省掉大量扯皮。
