1. 为什么要在IDEA里用Java写Spark词频统计很多人第一次接触Spark都是从Scala或者PySpark开始的这很正常——官方文档里Scala示例最多Python写起来也快。但实际工作中相当一部分团队的技术栈是Java。后端服务是Java写的构建工具是MavenCI流程里跑的是JDK你不可能为了一个数据处理任务单独维护一套Scala环境。所以用Java写Spark不是退而求其次而是很多生产环境的真实需求。词频统计这个例子看起来简单但它麻雀虽小五脏俱全涉及数据读取、转换、聚合、输出四个核心环节覆盖了Spark最基础的RDD操作链路。把它跑通你就有了一个可以往里填业务逻辑的骨架。后面不管是做日志分析、用户行为统计还是数据清洗套路都是在这个骨架上做替换。这篇文章面向的是有Java基础、但没怎么碰过Spark的开发者。我会从IDEA的环境配置讲起把依赖怎么选、代码怎么写、打包怎么打、本地跑和集群跑有什么区别一步步说清楚。中间会穿插一些我实际踩过的坑比如依赖冲突、序列化报错、日志刷屏这些问题这些在官方文档里往往一笔带过但实际开发中能卡你半天。需要说明的是Spark 3.2这个版本对Java 8和Java 11都支持但如果你用的是Java 17会遇到一些模块访问限制的问题这个后面会专门讲。另外IDEA的版本差异也会影响操作路径我尽量把不同版本的差异都标注出来。2. 环境搭建JDK、Maven与IDEA的版本搭配2.1 JDK版本的选择逻辑Spark 3.2官方编译目标是Java 8同时支持Java 11。如果你用Java 8基本不会遇到兼容性问题这是最稳妥的选择。Java 11也可以用但需要注意一些反射相关的警告。Java 17的话由于JDK内部模块的强封装Spark运行时需要额外添加JVM参数来开放模块访问权限否则会报InaccessibleObjectException。我的建议是如果这是一个新项目且团队没有强制要求直接用Java 8。不是因为它先进而是因为Spark生态里大量依赖比如Hadoop客户端、Kryo序列化库在Java 8上经过最充分的测试。用Java 11也行但你要做好偶尔处理兼容性问题的准备。在IDEA里配置JDK的路径是File → Project Structure → Project → SDK。这里选好之后还要确认Maven的编译插件也指向同一个版本否则会出现编译用Java 8、运行用Java 11的错位情况。2.2 Maven依赖的取舍Spark的Maven依赖有一个特点核心包和具体组件是分开的。spark-core只包含核心调度和RDD API如果你要用SQL或者Streaming需要单独引入对应的包。对于词频统计这个场景只需要spark-core就够了。dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.2.0/version /dependency注意artifactId后面的_2.12这表示它是用Scala 2.12编译的。Spark 3.2默认使用Scala 2.12如果你引入的是_2.13的包会出现类找不到的问题。这个后缀很容易被忽略但选错了就是一堆NoClassDefFoundError。还有一个关键点spark-core自带了对Hadoop客户端的依赖但版本可能和你环境里的不一致。如果你只是本地跑不涉及HDFS可以在依赖里排除掉Hadoop相关包减小打包体积dependency groupIdorg.apache.spark/groupId artifactIdspark-core_2.12/artifactId version3.2.0/version exclusions exclusion groupIdorg.apache.hadoop/groupId artifactId*/artifactId /exclusion /exclusions /dependency但如果你后续要读写HDFS就不能排除反而要确保Hadoop客户端版本和集群一致。这个取舍取决于你的实际场景。2.3 IDEA里容易忽略的两个配置第一个是Maven的自动导入。IDEA默认不会自动刷新依赖你改了pom.xml之后需要手动点一下刷新按钮或者开启自动导入Settings → Build → Build Tools → Maven → Importing → Automatically import。我见过不少新手改完pom之后直接跑代码然后报类找不到其实就是依赖没刷新。第二个是运行时的VM options。本地跑Spark的时候如果不加参数日志会刷满整个控制台根本看不到自己打印的结果。可以在Run Configuration里加上-Dspark.masterlocal[*]这样就不用在代码里硬编码setMaster了。另外建议把日志级别调一下在src/main/resources下放一个log4j2.properties文件rootLogger.level warn rootLogger.appenderRef.console.ref console这样控制台就清爽多了只看得到WARN以上的日志。3. 词频统计代码的逐层拆解3.1 从SparkContext的创建说起Spark 3.x里创建入口有两种方式SparkContext和SparkSession。如果你只做RDD操作SparkContext就够了如果要用DataFrame/Dataset API就需要SparkSession。词频统计用RDD实现更直观所以我们用SparkContext。SparkConf conf new SparkConf() .setAppName(WordCount) .setMaster(local[*]); SparkContext sc new SparkContext(conf);setMaster(local[*])表示用本地模式运行*表示使用所有可用的CPU核心。这里有个细节如果你在代码里硬编码了setMaster打包到集群上跑的时候会报错因为集群模式下master是由spark-submit指定的。所以更规范的做法是把master配置放在外部代码里不写本地测试时通过VM options传入。setAppName设置的应用名会显示在Spark Web UI上方便你区分不同的任务。名字起得有意义一点别用默认的SparkPi之类的。3.2 数据读取与分区的考量读取文本文件用textFile方法JavaRDDString lines sc.textFile(input/words.txt);这里传入的路径可以是本地文件路径也可以是HDFS路径。本地模式下如果传的是相对路径它相对于的是项目根目录IDEA里运行或者JVM的启动目录命令行运行这个差异经常导致文件找不到的问题。我的习惯是本地测试时用绝对路径避免歧义。textFile还有一个可选参数minPartitions用来指定最小分区数。默认情况下Spark会根据文件大小和HDFS块大小自动计算分区数。对于小文件可能只有1个分区这样并行度不够。你可以显式指定JavaRDDString lines sc.textFile(input/words.txt, 4);但也不是分区越多越好每个分区都有调度开销。一般来说分区数设为CPU核心数的2到4倍比较合适。3.3 flatMap与mapToPair的配合词频统计的核心转换逻辑是把每行文本拆成单词然后给每个单词打上计数1的标签。JavaRDDString words lines.flatMap(line - Arrays.asList(line.split( )).iterator()); JavaPairRDDString, Integer pairs words.mapToPair(word - new Tuple2(word, 1));flatMap和map的区别是map是一对一输入一行输出一个元素flatMap是一对多输入一行可以输出多个元素。这里用flatMap是因为一行文本包含多个单词。mapToPair是Java API特有的它返回的是JavaPairRDD也就是键值对形式的RDD。Scala里直接返回Tuple2就行Java里需要显式调用mapToPair来获得PairRDD因为后续的reduceByKey只能在PairRDD上调用。这里有个容易踩的坑line.split( )只按空格拆分如果文本里有多个连续空格、制表符或者标点符号会拆出空字符串。更健壮的写法是用正则line.split(\\s)但这样还是会把标点符号带进单词里。如果要处理真实文本还需要去掉标点、转小写等清洗步骤。词频统计的示例代码通常省略这些但实际项目中这些预处理是必不可少的。3.4 reduceByKey与groupByKey的选择聚合操作有两个选择reduceByKey和groupByKey。很多人直觉上会用groupByKey先把相同key的值收集到一起然后再求和。但这样做有一个严重的性能问题groupByKey会把所有相同key的值都传输到同一个节点上数据量大时会造成严重的内存压力和网络传输。reduceByKey则会在map端先做一次本地聚合combiner只把聚合后的结果传输到reduce端。对于词频统计这种求和场景两种方式结果一样但reduceByKey的性能要好得多。JavaPairRDDString, Integer counts pairs.reduceByKey((a, b) - a b);这个lambda表达式(a, b) - a b就是聚合函数a和b是同一个key的两个值返回它们的和。Spark会先在每个分区内做局部聚合再把不同分区的结果合并。记住一个原则能用reduceByKey就不要用groupByKey。除非你的逻辑确实需要拿到某个key的所有值比如求中位数否则reduceByKey永远是更优的选择。3.5 结果输出与触发执行的时机Spark的转换操作transformation是惰性的只有遇到行动操作action才会真正执行。reduceByKey是转换操作它不会立即计算。你需要调用一个action来触发ListTuple2String, Integer result counts.collect();collect()会把所有结果拉回到Driver端适合结果集较小的场景。如果结果很大用collect()会导致Driver内存溢出。更安全的做法是保存到文件counts.saveAsTextFile(output/wordcount);注意saveAsTextFile要求输出路径不存在如果目录已存在会报错。这是Spark的一个设计选择目的是避免误覆盖已有数据。每次运行前需要手动删除输出目录或者在代码里做判断。另外saveAsTextFile输出的结果可能分成多个文件每个分区一个文件名类似part-00000、part-00001。如果你想要单个文件可以在保存前用coalesce(1)合并分区但这样会牺牲并行度数据量大时不建议。4. 本地运行与集群提交的差异处理4.1 本地模式下的调试技巧本地模式最大的好处是可以直接断点调试。在IDEA里你可以在lambda表达式里打断点观察每个元素的处理过程。但要注意lambda里的断点可能会被多次触发每个元素一次调试时要有心理准备。另一个技巧是利用Spark Web UI。本地模式下默认会在4040端口启动一个Web界面你可以打开http://localhost:4040查看任务执行情况包括Stage划分、分区数、Shuffle数据量等。这些信息对理解Spark的执行机制非常有帮助。如果你在代码里用了Thread.sleep来观察UI记得在提交集群前去掉否则会白白浪费时间。4.2 打包时最容易出问题的环节用Maven打包Spark应用最常见的问题是依赖冲突。Spark本身依赖了大量的库Jackson、Guava、Netty等如果你的项目里也引入了这些库的不同版本运行时就会出现NoSuchMethodError或ClassNotFoundException。解决思路有两个一是用maven-shade-plugin打一个fat jar把所有依赖打进去但要注意排除Spark自身的包因为集群上已经有了二是只打业务代码依赖由spark-submit通过--jars参数指定。我倾向于第一种方式因为管理起来简单。配置大概是这样plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration filters filter artifact*:*/artifact excludes excludeMETA-INF/*.SF/exclude excludeMETA-INF/*.DSA/exclude excludeMETA-INF/*.RSA/exclude /excludes /filter /filters /configuration /execution /executions /plugin那几个META-INF的排除是必须的否则打出来的jar会有签名验证问题运行时报Invalid signature file digest。4.3 序列化问题的排查思路Spark在Shuffle过程中需要对数据进行序列化。Java的默认序列化机制效率不高Spark默认使用Kryo序列化。但Kryo要求注册自定义类如果你在RDD里用了自定义对象可能会遇到KryoException。词频统计这个例子里用的都是String和Integer不会遇到这个问题。但如果你后续扩展功能比如把单词封装成自定义的Word对象就需要注册conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer); conf.registerKryoClasses(new Class[]{Word.class});另一个常见的序列化问题是lambda表达式捕获了外部变量。如果捕获的变量没有实现Serializable接口运行时会报NotSerializableException。解决办法是把变量声明为final或者用静态变量或者实现Serializable接口。5. 从词频统计延伸出的实战经验5.1 中文分词的接入方式英文词频统计按空格拆分就行中文就麻烦了因为中文词与词之间没有空格。实际项目中需要接入分词库比如HanLP或者jieba的Java版本。接入方式是在flatMap里调用分词方法JavaRDDString words lines.flatMap(line - { ListString segList HanLP.segment(line); return segList.stream().map(term - term.word).iterator(); });但要注意分词库通常需要加载词典这个初始化过程比较耗时。如果放在flatMap里每次都初始化性能会很差。正确的做法是用mapPartitions在每个分区开始时初始化一次JavaRDDString words lines.mapPartitions(iter - { // 每个分区初始化一次 Segment segment HanLP.newSegment(); ListString result new ArrayList(); while (iter.hasNext()) { result.addAll(segment.seg(iter.next())); } return result.iterator(); });这个模式在处理需要初始化资源的场景比如连接数据库、加载模型时非常有用值得记住。5.2 数据倾斜的识别与缓解词频统计看起来简单但如果某个词出现的频率极高比如的、是这类停用词就会导致数据倾斜——某个分区的数据量远大于其他分区拖慢整个任务。识别数据倾斜的方法是看Spark Web UI里各个Task的处理时间如果某个Task明显比其他慢很多基本就是倾斜了。缓解办法有几种一是提前过滤停用词这是最简单的二是给key加随机前缀打散聚合后再去掉前缀二次聚合三是用reduceByKey的并行度参数增加分区数。对于词频统计最实用的还是过滤停用词。维护一个停用词表在分词后过滤掉SetString stopWords new HashSet(Arrays.asList(的, 是, 在, 了)); JavaRDDString filtered words.filter(word - !stopWords.contains(word));5.3 输出格式的控制saveAsTextFile默认输出的格式是(key, value)的元组形式带括号。如果你想要纯文本格式比如word\tcount需要先转换counts.map(t - t._1 \t t._2).saveAsTextFile(output);如果结果集不大也可以直接collect后在Driver端用Java的文件IO写出这样格式完全可控。但要注意collect的内存限制结果超过几百MB就不适合了。另外如果后续要用Excel或者数据库导入可以考虑输出CSV格式注意处理特殊字符的转义。5.4 资源参数的调优方向本地模式下资源参数意义不大但提交到集群时就需要关注了。几个关键参数参数含义建议值executor-memory每个Executor的内存根据数据量一般4G-8Gexecutor-cores每个Executor的CPU核心数2-4num-executorsExecutor数量根据集群规模driver-memoryDriver内存结果集大时调大这些参数没有万能公式需要根据数据量和集群资源反复测试。一个经验法则是先给一个保守的值跑一遍看Spark UI里的内存使用和GC情况再逐步调整。注意executor-memory不要超过节点物理内存的75%要给操作系统和其他进程留出空间。另外如果用了collect把大量数据拉回Driverdriver-memory要相应调大。6. 那些文档里不会写的踩坑记录6.1 日志框架冲突导致的启动失败Spark内部用的是log4j如果你的项目里引入了logback或者slf4j的其他实现可能会出现日志框架冲突表现为启动时没有任何输出或者报ClassNotFoundException。解决办法是在pom里排除掉冲突的日志依赖或者统一用log4j2。我遇到过一次项目里引入了spring-boot-starter它自带logback结果Spark任务启动就卡住。排查了半天才发现是日志框架的问题。后来在pom里加了排除exclusion groupIdch.qos.logback/groupId artifactIdlogback-classic/artifactId /exclusion6.2 Windows环境下Hadoop依赖的缺失在Windows上跑Spark如果代码里涉及文件读写可能会报Could not locate winutils.exe的错误。这是因为Hadoop在Windows上需要一些原生工具。解决办法是下载对应版本的winutils放到某个目录下然后设置环境变量HADOOP_HOMED:\hadoop并在代码里设置System.setProperty(hadoop.home.dir, D:\\hadoop);这个问题在Linux和Mac上不会遇到但Windows开发者几乎都会踩一次。6.3 IDEA运行时的类路径问题IDEA运行Java应用时类路径的构建方式和Maven打包后不一样。有时候Maven打包能跑但IDEA里直接运行就报类找不到。这通常是因为IDEA没有把某些依赖加入运行时的classpath。检查方法是File → Project Structure → Modules → Dependencies看看有没有标红的依赖。另一个常见问题是provided作用域的依赖。如果你把Spark依赖设为provided集群提交时的推荐做法IDEA运行时不会把它加入classpath导致本地跑不起来。解决办法是本地测试时临时改成compile打包时再改回provided或者用Maven的profile来管理。6.4 输出目录已存在的处理前面提到saveAsTextFile要求输出目录不存在。每次调试都要手动删目录很烦可以在代码里加一段判断Configuration hadoopConf new Configuration(); Path outputPath new Path(output/wordcount); FileSystem fs FileSystem.get(hadoopConf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); }但这段代码引入了Hadoop的FileSystem API如果你之前排除了Hadoop依赖这里会报错。所以取舍要看你的实际需求。另一个办法是用时间戳作为输出目录名每次运行都生成新目录避免冲突。6.5 并行度设置不当导致的性能问题默认情况下textFile读取的文件如果很小可能只有1个分区后续所有操作都是单线程的完全发挥不了Spark的并行优势。我见过有人跑一个几MB的文件等了半天以为是Spark慢其实是并行度没设对。解决办法是在读取时指定分区数或者在聚合前用repartition重新分区JavaRDDString lines sc.textFile(input/words.txt).repartition(8);但repartition会触发Shuffle有一定开销。如果数据源本身就是多个文件Spark会自动为每个文件创建分区这种情况就不需要手动干预了。6.6 内存溢出时的排查顺序Spark任务报OutOfMemoryError时先别急着加内存按这个顺序排查第一看是Driver还是Executor溢出。Driver溢出通常是collect了大量数据或者broadcast的变量太大。Executor溢出通常是单个分区的数据量过大或者聚合操作的内存占用过高。第二看Spark UI里的Storage和Executor页面确认内存使用分布。如果某个Executor的内存持续增长然后溢出很可能是数据倾斜。第三考虑调整序列化方式。Kryo比Java序列化省内存如果还没用Kryo可以先切换试试。第四才是调整内存参数。加内存能解决一部分问题但如果是数据倾斜导致的加内存只是延缓溢出时间治标不治本。这套排查顺序是我踩了很多次坑之后总结出来的先定位问题类型再针对性解决比盲目调参高效得多。