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 内部类。下面这张表是我实测下来比较稳的组合：

选 Scala 2.12 而不是 2.13，是因为大部分生产环境的 Hadoop 生态还是 2.12 为主，2.13 的包相对少。JDK 方面，如果你非要用 17，记得在启动参数里加上--add-opens java.base/java.lang=ALL-UNNAMED这类开放模块的配置，否则 Spark 的序列化会直接报错。

2.2 Maven 依赖的取舍与排除

依赖配置我直接给一份可用的骨架，重点看exclusions部分：

<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.12</artifactId> <version>3.2.4</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> <exclusion> <groupId>log4j</groupId> <artifactId>log4j</artifactId> </exclusion> <exclusion> <groupId>org.codehaus.janino</groupId> <artifactId>janino</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.2.4</version> </dependency>

为什么要排除janino？因为 Spring Boot 的spring-boot-devtools或某些模板引擎也会引入 janino，版本不一致时会导致 Spark SQL 的代码生成失败。排除后由 Spring Boot 统一管理版本，问题就消失了。

另外，spark-sql是必须的，因为实际业务里没人只用 RDD，DataFrame 和 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.MyKryoRegistrator

master在 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是重量级对象，创建一次要几秒钟。如果你的服务是多实例部署，每个实例都会创建一个 SparkSession，local 模式下没问题，集群模式下会各自连集群，资源消耗要提前评估。

3.2 任务提交接口的设计

HTTP 接口我设计成三个：提交任务、查询状态、获取结果。提交接口用 POST，返回一个任务 ID：

@RestController @RequestMapping("/api/spark") public class SparkJobController { @Autowired private SparkJobService sparkJobService; @PostMapping("/submit") public ResponseEntity<String> submit(@RequestBody JobRequest request) { String jobId = sparkJobService.submitJob(request); return ResponseEntity.ok(jobId); } @GetMapping("/status/{jobId}") public ResponseEntity<JobStatus> status(@PathVariable String jobId) { return ResponseEntity.ok(sparkJobService.getStatus(jobId)); } @GetMapping("/result/{jobId}") public ResponseEntity<List<Map<String, 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 List<Map<String, Object>> result; }

结果缓存用ConcurrentHashMap就够了，但如果任务量大，建议加过期清理策略，比如任务完成 1 小时后自动移除。我试过用 Caffeine 做本地缓存，效果不错，配置expireAfterWrite(1, TimeUnit.HOURS)即可。

注意：结果集不要直接存 DataFrame，DataFrame 持有 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 到 20，1G 到 10G 设 50 到 100，10G 以上再考虑 200 以上。

第二个优化点是缓存策略。如果同一个 DataFrame 被多次使用，一定要cache()或persist()。但不要滥用，缓存会占内存，用完后unpersist()释放。我见过有人把所有 DataFrame 都 cache，结果 executor OOM。

第三个优化点是广播变量。小表 join 大表时，用broadcast()把小表广播出去，避免 shuffle。这个优化效果非常明显，我实测过一个 10G 大表 join 100M 小表的场景，广播后耗时从 8 分钟降到 1 分钟。

Dataset<Row> 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.ch=ALL-UNNAMED --add-opens java.base/java.lang=ALL-UNNAMED --add-opens java.base/java.util=ALL-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 space

driver OOM 一般是collect()数据量太大，改成take()或写文件。executor OOM 一般是分区数据不均匀或缓存太多，调整分区数或减少缓存。

问题六：HTTP 请求超时

Spark 任务执行时间超过网关的超时限制。解决办法是改成异步提交，前端轮询状态。如果必须同步，把网关超时调到 10 分钟以上，但不推荐。

5.3 常见问题速查表

5.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 name="org.apache.spark" level="WARN"/> <logger name="org.apache.hadoop" level="WARN"/>

第四个技巧：本地模式下的临时目录。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 Map<String, 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 { List<Map<String, 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 List<Map<String, Object>> executeJob(JobRequest request) { Dataset<Row> df = sparkSession.read() .option("header", "true") .csv(request.getInputPath()); Dataset<Row> 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 List<Map<String, 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。

调优后总耗时从 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就能看到任务执行详情，比看日志直观得多。这个习惯帮我省了大量排查时间。