Spark 4.1.2 + Scala 2.13 开发环境：IntelliJ IDEA 2024.3 与 SBT 1.9.9 配置实战

Spark 4.1.2 Scala 2.13 开发环境IntelliJ IDEA 2024.3 与 SBT 1.9.9 配置实战对于希望构建现代化Spark应用的开发者而言搭建一个高效的本地开发环境至关重要。本文将详细介绍如何在IntelliJ IDEA 2024.3中配置Spark 4.1.2与Scala 2.13的开发环境使用SBT 1.9.9作为构建工具并提供完整的项目配置示例。1. 环境准备与工具安装在开始Spark项目开发前需要确保系统中已安装以下基础组件Java Development Kit (JDK)Spark 4.1.2要求JDK 8或更高版本推荐使用JDK 11或17以获得最佳兼容性Scala 2.13与Spark 4.1.2兼容的Scala版本IntelliJ IDEA 2024.3安装时需包含Scala插件SBT 1.9.9Scala的构建工具1.1 JDK安装验证在终端中运行以下命令检查Java版本java -version预期输出应类似于openjdk version 17.0.3 2022-04-19 OpenJDK Runtime Environment (build 17.0.37) OpenJDK 64-Bit Server VM (build 17.0.37, mixed mode)1.2 Scala环境配置使用SDKMAN!安装特定版本的Scalasdk install scala 2.13.12安装后验证版本scala -version正确配置应显示Scala code runner version 2.13.12 -- Copyright 2002-2023, LAMP/EPFL and Lightbend, Inc.2. IntelliJ IDEA项目初始化2.1 创建新SBT项目打开IntelliJ IDEA选择New Project在左侧菜单中选择Scala右侧选择SBT设置项目名称和位置确保SDK选择正确的JDK版本如JDK 17SBT版本指定为1.9.9Scala版本选择2.13.12点击Create完成项目创建2.2 基础项目结构创建完成后项目应具有以下目录结构my-spark-project/ ├── build.sbt ├── project/ │ ├── build.properties │ └── plugins.sbt ├── src/ │ ├── main/ │ │ ├── resources/ │ │ └── scala/ │ └── test/ │ ├── resources/ │ └── scala/ └── target/3. 配置build.sbt文件build.sbt是SBT项目的核心配置文件。以下是Spark 4.1.2与Scala 2.13的完整配置示例ThisBuild / version : 1.0.0 ThisBuild / scalaVersion : 2.13.12 ThisBuild / organization : com.example // 解决Java模块系统在Spark中的警告 ThisBuild / run / fork : true ThisBuild / run / javaOptions --add-opensjava.base/sun.nio.chALL-UNNAMED lazy val root (project in file(.)) .settings( name : spark-scala-demo, libraryDependencies Seq( // Spark核心依赖 org.apache.spark %% spark-core % 4.1.2, org.apache.spark %% spark-sql % 4.1.2, // 测试依赖 org.scalatest %% scalatest % 3.2.16 % Test, // 日志处理 org.apache.logging.log4j % log4j-api % 2.20.0, org.apache.logging.log4j % log4j-core % 2.20.0, org.apache.logging.log4j % log4j-slf4j-impl % 2.20.0 ), // 并行执行设置 parallelExecution in Test : false, // 编译器选项 scalacOptions Seq( -deprecation, -feature, -unchecked, -Xlint, -Ywarn-dead-code, -Ywarn-numeric-widen, -Ywarn-unused, -Ywarn-value-discard ), // JVM运行参数 javaOptions Seq( -Xms512M, -Xmx2G, -XX:UseG1GC, -Dlog4j.configurationFilelog4j2.properties ) )3.1 关键配置说明Spark依赖管理spark-core和spark-sql是Spark的基础模块使用%%确保获取与Scala 2.13兼容的版本日志配置添加Log4j2依赖以避免Spark的日志冲突需要在src/main/resources下添加log4j2.properties文件JVM参数设置合理的堆内存大小-Xms和-Xmx使用G1垃圾收集器提高性能4. 解决常见依赖冲突Spark项目经常会遇到依赖冲突问题以下是两个典型场景的解决方案4.1 Guava版本冲突Spark 4.1.2依赖Guava 14.0.1而许多现代库需要更高版本。解决方案dependencyOverrides com.google.guava % guava % 32.1.3-jre4.2 Jackson版本冲突Spark自带Jackson模块可能与用户引入的版本冲突dependencyOverrides Seq( com.fasterxml.jackson.core % jackson-core % 2.15.3, com.fasterxml.jackson.core % jackson-databind % 2.15.3, com.fasterxml.jackson.module %% jackson-module-scala % 2.15.3 )提示使用sbt dependencyTree命令可以查看完整的依赖关系树帮助识别冲突5. 编写第一个Spark应用在src/main/scala目录下创建SimpleApp.scala文件import org.apache.spark.sql.SparkSession object SimpleApp { def main(args: Array[String]): Unit { // 创建SparkSession val spark SparkSession.builder() .appName(Simple Application) .master(local[*]) // 使用本地模式所有可用核心 .config(spark.sql.shuffle.partitions, 4) // 优化小数据集性能 .getOrCreate() // 减少日志输出 spark.sparkContext.setLogLevel(WARN) import spark.implicits._ try { // 创建示例DataFrame val data Seq( (Java, 20000), (Python, 100000), (Scala, 3000) ) val df data.toDF(language, users) // 执行简单转换 val filteredDF df.filter($users 10000) // 显示结果 filteredDF.show() // 执行聚合操作 df.createOrReplaceTempView(languages) val resultDF spark.sql( |SELECT language, users |FROM languages |WHERE users (SELECT AVG(users) FROM languages) |.stripMargin) resultDF.show() } finally { // 确保SparkSession被正确关闭 spark.stop() } } }5.1 运行配置在IntelliJ中右键点击SimpleApp对象选择Run SimpleApp或创建自定义运行配置Main class:SimpleAppVM options:-Xmx2G -Dlog4j.configurationFilelog4j2.propertiesWorking directory: 项目根目录Environment variables: 可添加SPARK_LOCAL_IP127.0.0.16. 性能优化配置在build.sbt中添加以下配置可显著提升本地开发体验// 针对本地开发的优化配置 fork in run : true outputStrategy : Some(StdoutOutput) connectInput in run : true // 内存设置 javaOptions in run Seq( -Xms1G, -Xmx4G, -XX:MaxMetaspaceSize1G, -XX:UseG1GC, -XX:MaxGCPauseMillis200, -Djava.net.preferIPv4Stacktrue ) // 针对Spark的特定配置 javaOptions in run Seq( -Dspark.masterlocal[*], -Dspark.driver.memory2G, -Dspark.executor.memory2G, -Dspark.ui.enabledtrue, -Dspark.ui.port4040 )6.1 日志配置示例在src/main/resources/log4j2.properties中添加rootLogger.level WARN rootLogger.appenderRef.stdout.ref console appender.console.type Console appender.console.name console appender.console.layout.type PatternLayout appender.console.layout.pattern %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n logger.spark.name org.apache.spark logger.spark.level WARN logger.jetty.name org.sparkproject.jetty logger.jetty.level WARN7. 高级项目配置7.1 多模块项目结构对于复杂项目可以拆分为多个模块lazy val core (project in file(core)) .settings( name : spark-core, // 核心配置 ) lazy val utils (project in file(utils)) .settings( name : spark-utils, // 工具类配置 ) lazy val app (project in file(app)) .dependsOn(core, utils) .settings( name : spark-app, // 应用配置 )7.2 打包配置添加assembly插件用于创建可部署的fat JAR在project/plugins.sbt中添加addSbtPlugin(com.eed3si9n % sbt-assembly % 2.1.1)在build.sbt中添加合并策略assembly / assemblyMergeStrategy : { case PathList(META-INF, services, xs _*) MergeStrategy.filterDistinctLines case PathList(META-INF, xs _*) MergeStrategy.discard case reference.conf MergeStrategy.concat case _ MergeStrategy.first }使用命令创建部署包sbt assembly8. 实际开发技巧8.1 使用Spark Shell进行快速验证在SBT项目中集成Spark Shell功能lazy val sparkShell taskKey[Unit](Start Spark shell with project dependencies) sparkShell : { val sbtClasspath (fullClasspath in Runtime).value.files.absString val cmd sspark-shell --jars $sbtClasspath println(sRunning: $cmd) cmd ! }运行方式sbt sparkShell8.2 性能监控配置在spark-defaults.conf中添加位于src/main/resourcesspark.eventLog.enabled true spark.eventLog.dir file:///tmp/spark-events spark.history.fs.logDirectory file:///tmp/spark-events spark.metrics.conf.*.sink.console.class org.apache.spark.metrics.sink.ConsoleSink8.3 单元测试示例在src/test/scala中创建测试类import org.apache.spark.sql.SparkSession import org.scalatest.BeforeAndAfterAll import org.scalatest.funsuite.AnyFunSuite class SparkTestExample extends AnyFunSuite with BeforeAndAfterAll { private var spark: SparkSession _ override def beforeAll(): Unit { spark SparkSession.builder() .appName(Testing) .master(local[2]) .config(spark.ui.enabled, false) .getOrCreate() } override def afterAll(): Unit { if (spark ! null) { spark.stop() } } test(simple dataframe creation) { import spark.implicits._ val data Seq((a, 1), (b, 2)) val df data.toDF(letter, number) assert(df.count() 2) assert(df.columns.length 2) } }运行测试sbt test