很多做数据开发的朋友第一次接触 Spark 时都会被 Scala 的函数式写法弄得有点懵。明明用 Python 写 Pandas 挺顺手一到 Spark 里要用 Scala 写算子map、flatMap、reduceByKey一串串冒出来语法和 Java 相差很大和 Python 也不是一个路数。但如果你打算认真做 Spark 开发Scala 几乎是个绕不开的选择——它才是 Spark 的原生语言很多新特性都是 Scala 版先上而且 Scala 在类型安全、算子表达力上确实有不可替代的优势。这篇就把我在实际开发中高频用到的 Spark/Scala 语法整理一遍当作一份可以随时翻阅的实战手册。1. 为什么 Spark 开发绕不开 Scala1.1 Scala 与 Spark 的深度绑定不是玄学先说结论你完全可以用 PySpark 做数据开发但如果涉及性能调优、自定义算子、读 Spark 源码Scala 的优先级会越来越高。Spark 本身就是用 Scala 写的这意味着它的核心抽象——RDD、DStream、Dataset——在设计之初就是围绕 Scala 的函数式编程特性构建的。默认支持 lambda 表达式、隐式转换、case class 自动序列化这些在 Java 里反反复复写样板代码才能实现的功能Scala 几行就搞定了。我在实际项目里遇到过一种典型情况某团队用 PySpark 做实时特征计算每个任务的数据量在百万级但 PySpark 里使用 UDF 时要经历 Python 和 JVM 之间的序列化通信性能损耗非常明显。后来同一套逻辑改用 Scala 实现相同的资源消耗下整体时间压缩了百分之三十左右。原因并不神秘——Scala 代码直接运行在 JVM 里算子的执行路径短省掉了一层 Python 解释和 Py4J 通信的开销。如果数据量小这点差异不明显但上了生产环境千万级、亿级数据时体感就完全不同了。1.2 函数式思维才是 Spark 的核心语法书只教你皮毛很多人学 Scala 语法翻了几页官网文档发现变量声明、循环、类这些都和 Java 差不了太多然后就开始写 Spark 作业了。但实际上Scala 用在 Spark 里最关键的并不是类怎么写这类 OOP 语法而是函数式编程的那套思维把数据当作不可变集合用变换操作代替循环用函数作为参数传递行为。举个例子刚入门的人写清洗逻辑脑子里的第一反应是这样的val result new ArrayBuffer[String]() for (line - lines) { if (line.contains(error)) { result line.trim.toLowerCase } }这个写法在普通 Scala 程序里没毛病但在 Spark 里不太对路。你应该写的是val result lines .filter(_.contains(error)) .map(_.trim.toLowerCase)两者的计算结果一样但风格完全不同。后者是声明式的描述我要从这里面筛出错误行、做清理而不是一步步指挥机器先把数组建出来再遍历再判断再塞进新数组。Spark 的算子本质上是把后一种风格抽象成了filter、map这些高阶函数。理解了这一点Spark 的语法学习其实就从背 API变成了理解数据流的变换节奏。1.3 生产环境到底选 Scala 还是 PySpark 或者 Java我在不同公司见过不同的技术选型这里把个人经验写出来供参考。数据量较小、团队以 Python 为主、业务迭代速度要求特别高的场景PySpark 确实开发效率高。但任务一旦进入定时调度、需要对接公司内部统一的数据平台、要求稳定的运行性能和可控的依赖Scala 的优势就会显露出来。Java 也能写 Spark但表达同样逻辑需要的代码量大概是 Scala 的 2 到 3 倍而且算子传函数时没有 Scala 那么自然。我见过一个 Java 项目里为了传一个简单的过滤函数不得不写匿名内部类代码读起来相当痛苦。所以如果你打算长期深耕大数据开发我的建议很直接不要把 Scala 看成另外一种语言把它看成 Spark 的配套基础设施。不需要学到多高深但高频语法点必须熟练到条件反射。2. 高频基础语法与函数式写法先过一遍2.1 val、var、不可变集合这决定了你的算子是否安全Scala 里的变量分两种val和var。val是不可变的一旦赋值就不能再指到别的地方var是可变的可以重复赋值。在 Spark 的分布式执行环境下强烈建议全部使用val。原因在于当某个变量被算子捕获并分发到多个 Executor 上执行时如果它是不可变的就不存在并发修改的问题整个作业的确定性会强很多。另外注意区分引用不可变和指向的对象内部可变这两个概念。val list ArrayBuffer(1,2,3)这种写法虽然list这个引用不能变但ArrayBuffer内部的元素是可以增删的。放在 Spark 的算子外面可能没问题但一旦在算子里面对共享的可变对象做修改几乎必然是踩并发坑的前兆。尽量用List、Seq、Vector这类不可变集合操作返回的都是新集合不会污染原数据。2.2 集合操作三件套map、filter、flatMap以及它们和 for 推导式的等价关系Scala 集合里最高频的三个方法就是map、filter、flatMap。map对每个元素做一对一映射filter按条件筛掉元素flatMap是先映射后展平——也就是每个元素经过函数处理后得到的是一个集合最终结果把所有集合并在一起。Spark 里这三个方法的使用频率同样极高但注意集合上的map和 Spark RDD 的map不是同一个东西它们只是理念一致。用 Scala 集合练手时写过List(1,2,3).map(_ * 2)到 Spark 里对 RDD 写rdd.map(_ * 2)时思路是一模一样的只是执行引擎不同而已。顺带提一下 for 推导式。Scala 里很多集合操作用 for 也能写例如val doubled for (x - list) yield x * 2和list.map(_ * 2)是等价的。但 Spark 算子内部不支持这种语法来操作分布式数据集所以实际写 Spark 作业时还是直接使用算子更顺手。不过理解 for 推导式有助于读懂 Scala 项目里别人写的老代码。2.3 函数字面量与占位符下划线就是那个高频配角Scala 函数的写法在 Spark 里随处可见。最完整的形式是(x: Int) x * 2但因为有类型推断很多时候可以简化成x x * 2。再往下还有一种极简写法就是占位符_。_在 Scala 里承担了多种角色初学者经常被搞晕。最核心的是三种用法作为函数参数的占位符list.map(_ * 2)等价于list.map(x x * 2)。作为通配符导入包import spark.implicits._意思就是把这个包里的所有隐含成员都导入。作为默认值初始化var name: String _先声明一个空引用后续再赋值。在 Spark 作业里你看到大量_.开头的表达式比如rdd.filter(_.startsWith(ERROR))这里的_指的就是集合中的每个元素。熟练之后这种写法比写完整 lambda 快很多。但要注意一点如果函数参数出现两次_就不适用了。比如(a, b) a b你不能写成_ _之外的形式吗其实_ _是可以的两个占位符分别代表第一个和第二个参数。那(a, b) a b简化成_ _是正确的。但如果参数用法更复杂比如(a, b) a b * 2就直接用完整 lambda别强行套用占位符。2.4 case classSpark 里的 Schema 担当Scala 里定义一个case class极其方便case class User(id: Long, name: String, age: Int)这个简单声明带来的好处是自动生成equals、hashCode、toString、copy方法并且可以直接从构造参数反向提取字段。这意味着你在写模式匹配时可以直接解构对象。在 Spark 里case class 最重要的作用是用作 DataFrame 的 Schema。比如你在读一个 CSV 文件可以定义一个 case class 对应每一行然后val ds spark.read.csv(path).as[User]这样后续操作里ds就是一个带类型的Dataset[User]每行都是强类型的User对象写ds.filter(_.age 18)比用字符串列名优雅得多而且编译期就能发现字段名错误。这在大型作业里是非常实用的防呆手段。我强烈建议每个 Spark 项目里的数据模型层都用 case class 来定义。即使最开始只是做简单 ETL后面一旦需要复用、加字段、做类型转换case class 的优势会越来越明显。2.5 伴生对象与 apply 方法的实际作用在 Scala 里object User被称为class User的伴生对象。两者可以互相访问私有成员。case class 自动生成伴生对象里面默认带了一个apply方法所以你可以直接User(1, Alice, 20)而不用写new User(...)。在 Spark 作业里伴生对象最常见的用途是放置常量、工具函数和隐式转换。例如object DataConstants { val MaxRetry 3 val ErrorCode 500 }然后在节点代码里直接DataConstants.MaxRetry引用。这比在类内部塞一堆静态常量要清晰得多也更符合 Scala 的代码组织习惯。3. Spark 核心算子与常用语法实战RDD 和 PairRDD3.1 RDD 的 map、filter、flatMap 与 mapPartitions 的性能差异前面讲了集合层面的map、filter、flatMap到了 Spark 的 RDD 上这些算子的名字和作用基本延续下来了但执行层完全不同。RDD 的map是逐元素处理每条数据经过函数后返回一条新数据。在分布式环境下一个分区里的数据可能有很多条map就是一条一条来的每一条都走一遍函数。日常写 ETL 时mapPartitions是一个非常值得关注的算子。它的参数是一个函数作用在一个分区上一次返回整体结果而不是对单个元素处理。比如你要对一个分区内的所有记录做批量初始化操作或者需要批量发送一个外部请求在map里每条记录都建立连接就太浪费了在mapPartitions里每个分区建一次连接、批量处理一批数据性能能提升不少。但这里有个坑mapPartitions函数必须返回一个迭代器而且整个分区的数据会一次性交给你的函数。如果你的分区太大处理函数里产生了过大的中间结果就容易内存溢出。所以用mapPartitions时分区大小控制和函数内部的资源释放逻辑都要好好设计。3.2 reduceByKey 与 groupByKey выбор 背后的性能和正确性逻辑处理 PairRDD键值对 RDD时reduceByKey和groupByKey是出现频率最高的两个。它们都能把相同 key 的数据聚合到一起但执行方式差别很大。groupByKey的逻辑是先把所有相同 key 的 value 收集到一个迭代器里然后再一起处理。这里有几个明显的问题数据在网络中洗牌时value 是全部传输的如果某个 key 对应的 value 特别多单机内存压力会很大。reduceByKey则是在洗牌之前先在每个分区内对相同 key 做一次预聚合然后再洗牌合并。这个过程叫 map 端本地聚合能显著减少网络传输和最终节点的数据处理量。在实际项目中能选reduceByKey就不要选groupByKey。比如计算每个用户的访问次数val visits: RDD[(String, Int)] ... val count visits.reduceByKey(_ _)等价逻辑用groupByKey写是这样val count visits.groupByKey().mapValues(_.sum)功能一致但上一行比下一行在大量数据下快很多。原因就是 reduceByKey 提前做了一次局部求和。不过如果聚合逻辑不是简单的求和、不能定义局部聚合函数比如你要把同一个 key 的所有消息体按时间排序就必须用groupByKey。这种场景下你能做的优化是尽量调整分区器让相同 key 的数据尽量落在同一个分区减少跨节点传输。3.3 join 系列join、leftOuterJoin、cogroup关联逻辑的常用选择PairRDD 之间的 join 操作非常通用。rdd1.join(rdd2)会取两个 RDD 中 key 都存在的记录做内连接。leftOuterJoin保留左表全部 key右表匹配不上时 value 部分为None。在实际开发中我一般用cogroup的场景偏多尤其在需要基于同 key 做复杂关联逻辑的时候。cogroup会把两个 RDD 中相同 key 的两组 value 分别聚合成两个迭代器然后给你一个机会同时对它们做处理。它可以方便地实现逻辑复杂的关联比如左表存在但右表不存在时要生成特殊标记这种逻辑用join加filter也能做但代码会绕很多。举个例子val left: RDD[(String, Int)] ... val right: RDD[(String, String)] ... val result left.cogroup(right).map { case (key, (leftValues, rightValues)) // leftValues 是所有左表 value 的迭代器 // rightValues 是所有右表 value 的迭代器 ... }这种方式非常灵活也能在同一个地方处理多种关联逻辑。3.4 函数序列化问题写算子时最容易踩的坑Spark 的算子函数会被分发到多个 Executor 上执行。你的函数里如果引用了外部对象那个对象也必须能被序列化通过网络传给 Executor。最常见的报错是Task not serializable多半是算子里用了不可序列化的类。解决思路如下如果引用的是一个本身不可序列化的类实例尝试把它改写成只依赖基本类型和字符串的纯函数。如果引用的是某个配置对象用广播变量把配置传过去。在类内部写算子方法时把方法改成private val function: ... ...并通过transient修饰有时也能绕过去。我的习惯是所有算子里传的函数都写成纯函数只依赖函数入参和基本类型。这样从根上避开序列化难题。如果你的业务逻辑必须引用外部配置那就在 driver 端把配置序列化成一个 Map再通过广播变量分发。4. DataFrame 与 Dataset 的 DSL 语法比 SQL 更灵活的日常操作4.1 列引用方式col、$colName、colName、ds(colName)在 Spark SQL 和 DataFrame 的 DSL 操作里如何引用列是最基本的语法问题。常见方式有col(columnName)最标准的方式。$columnName需要引入spark.implicits._后使用最简洁。columnName使用了 Scala 的 Symbol 类型比较老式的写法。df.col(columnName)通过 DataFrame 对象引用。我的偏好是$columnName原因是写起来干净阅读时也能一眼看出是列引用。但在复杂的长表达式里col(columnName)会更清楚因为$符号容易和字符串插值混淆。这纯粹是团队风格问题没有绝对优劣。实际使用时要注意字符串形式的列名如果包含空格、点号或特殊符号引用方式要调整。比如列名是user.name直接用$user.name会被解析为嵌套字段访问而不是一个叫user.name的列。此时需要用col(user.name)这种方式把整个列名包起来或者把 DataFrame 的列改名尽量减少特殊字符。4.2 select、withColumn、filter 与 when 的组合用法日常做 ETL用得最多的就是select、withColumn、filter这三个方法。select负责投影选出指定的列也可以同时做列表达式操作val result df.select( $id, $name, ($age 1).as(next_age) )withColumn是新增一列或替换一列最方便的地方在于它不用把所有列重写一遍。withColumn其实可以理解成select的语法糖。新增列时要注意命名冲突如果新列名已经存在默认会覆盖掉原列这是一把双刃剑用好了改列值很方便用错了会把原始数据悄悄改掉。条件逻辑方面when和otherwise配合使用非常自然就像 SQL 里的CASE WHENval tagged df.withColumn( level, when($score 90, A) .when($score 60, B) .otherwise(C) )这种写法比UDF更高效因为when是 Spark 内置的逻辑表达式不走序列化开销。优先用内置 DSL实在搞不定了再上 UDF。4.3 groupBy、agg、cube、rollup多维统计的 DSL 写法groupBy加agg是分组聚合的基本句型df.groupBy($city) .agg( count(*).as(total), sum($amount).as(sum_amount), avg($amount).as(avg_amount) )这里注意agg里可以同时写多个聚合函数比一个个groupBy再 join 高效得多。cube和rollup则是数据仓库里多维分析常用的语法。rollup从左到右逐步汇总层级cube对所有维度组合做汇总。在 Scala DSL 里写法都很直接df.rollup($province, $city).agg(sum($amount)) df.cube($province, $city).agg(sum($amount))如果只是做基础报表groupByagg完全够用。只有到了分析型场景需要多维度小计时rollup和cube才值得拿出来用。4.4 UDF 与 UDAF什么时候该用、什么时候别用用户自定义函数UDF是扩展 Spark 表达能力的重要工具。定义方式很简单val toUpper udf((s: String) s.toUpperCase) df.withColumn(upper_name, toUpper($name))注意几个点UDF 的参数和返回值类型要尽量简单最好是基本类型和字符串如果传复杂对象序列化和类型匹配会非常麻烦。UDF 一旦注册Spark 无法对它做优化所以它的性能通常低于内置函数。能用when、concat、substring等内置函数解决的就不要用 UDF。UDAF用户自定义聚合函数更复杂一般用于自定义聚合逻辑。如果只需要sum、avg、collect_list内置函数都覆盖了。需要实现像分组内取中位数这种特殊逻辑时才需要自己定义 UDAF。实现 UDAF 在 Scala 里要继承Aggregator抽象类重写那几个方法。实际开发中我用到的频率不算高但一旦用到就非常关键。4.5 Dataset 的类型安全优势以及 Encoder 的隐含知识Dataset 是 DataFrame 的类型化版本。它提供了编译期类型检查能把很多运行时错误提前到编译时暴露。比如Dataset[User]你用.filter(_.age 18)就能在编译期确定age字段存在且类型为Int。要实现 Dataset 与 case class 之间的转换必须依赖Encoder。Spark 对常见类型和 Product 类型case class 属于 Product都有隐式 Encoder你只需要在代码里导入spark.implicits._然后import spark.implicits._ val ds: Dataset[User] df.as[User]如果 case class 里包含了自定义类型或者模板类型就可能出现找不到 Encoder的报错。解决办法是手动给自定义类型提供一个Encoder通常是利用Encoders.product或Encoders.bean组合出来。这里有个容易忽略的细节Dataset 的filter和 DataFrame 的filter行为不完全一样。DataFrame 的filter参数通常是 Column 表达式Dataset 的filter可以直接传一个 Scala 函数。这让你可以用强类型的方式处理数据代码更接近集合操作。5. 从语法到作业设计的进阶经验写出不那么脆弱的 Spark 作业5.1 常驻变量与广播变量分发配置的正确姿势Spark 作业里偶尔会有一个需求算子里要用到一份外部配置比如补全城市名对应的省份名。如果这份配置是 Map最直接的想法是把它定义在算子外部然后在map里直接引用。问题在于这个 Map 会在每个 Task 执行时被序列化和分发一次任务量一大传输开销成倍增长。正确的方案是把大配置用broadcast广播出去val cityMap: Map[String, String] ... val bc spark.sparkContext.broadcast(cityMap) val result df.as[User].map { user val province bc.value.getOrElse(user.city, 未知) ... }广播变量在集群中每个 Executor 上只保存一份副本多个 Task 共享读取大幅减少了网络传输和内存占用。注意广播变量是只读的你不能在 Executor 端修改它的内容。判断何时该用广播变量的经验是当引用的数据集足够小可以完全放进单机内存并且它在算子里被高频读取时就应该广播。反之如果数据集很大就应该考虑把它 join 进来而不是广播。5.2 累加器与日志收集简单但容易用歪的语法累加器Accumulator是 Spark 提供的分布式只写变量。Driver 定义一个初始值各 Executor 只能对它做累加操作最后由 Driver 读取结果。最常见的用途是统计数据清洗过程中过滤掉的异常记录数量。val badRecords spark.sparkContext.longAccumulator(bad) val cleaned df.filter { row val valid row.getAs[String](content) ! null if (!valid) bad.add(1) valid } println(sbad count: ${badRecords.value})这里有一个特别容易踩的坑如果累加器的累加动作放在不执行的算子里它的值不会更新。比如放在.filter之后但你并没有触发行动操作累计结果可能是 0。另外同一个算子如果执行多次累加器可能被重复累加。调试这类问题时要时刻记住累加器的结果只在任务真正执行完以后才可靠。5.3 缓存与持久化cache、persist、checkpoint 的取舍在迭代计算或者同一份 DataFrame 被多次复用的场景下缓存能省不少时间。cache()是将数据缓存在内存中persist(StorageLevel.DISK_ONLY)可以把数据放磁盘checkpoint则会切断 RDD 的血缘关系把中间结果存到可靠存储。cache()的默认存储级别是MEMORY_ONLY如果内存不足部分数据会被重新计算这不见得比不缓存更快。所以判断是否缓存时要结合内存资源和血缘链长度来考虑。一个经验法则是如果同一份数据会被多个行动操作用到且重新计算代价很高血缘链很长依赖数据巨大考虑缓存如果资源紧张用persist(StorageLevel.MEMORY_AND_DISK)让没放进内存的数据落盘。checkpoint 与 cache 不同它会打断血缘依赖。血缘太长的作业某些节点出错后需要回退重算整条链代价极高。用 checkpoint 把中间结果固化一下后续计算依赖的就是固化的数据。代价是会引入额外的 I/O 操作所以不能滥用。5.4 数据倾斜的基础应对从语法层面做一些努力数据倾斜是 Spark 作业最常见的性能杀手。表现在某个 key 的数据量特别大导致处理这个 key 的 Task 耗时远超其他 Task。从语法层面能做的优化思路有对于两表 join如果一张表很小尝试用广播 join彻底避免 shuffle。Scala 代码里可以显式提示df1.join(broadcast(df2), Seq(id))。对于 groupByKey 和 reduceByKey 聚合时遇到的倾斜可以考虑在 key 上添加随机前缀打散到多个 Task 先做一次局部聚合再去除前缀做第二次全局聚合。这个思路要用到两段式 map代码稍微复杂一点但效果明显。调整分区数通过repartition或coalesce让数据分布更均衡。coalesce只会减少分区数repartition会做一次全量 shuffle可能增加也可能减少分区。我自己的经验是遇到倾斜不要慌先用df.groupBy($key).count().orderBy($count.desc).show(10)看一眼数据分布判断是不是少数 key 造成的问题再决定用哪种手段。盲目调参数往往解决不了根本问题。5.5 分区数量的控制别让分区思维停留在默认值Spark 作业実行速度受分区数量影响很大。分区太多每个 Task 处理的平均数据量小调度和序列化开销反而变大分区太少每个 Task 数据量过大容易出现内存溢出和 GC 问题。一个实用的起点是让每个分区处理的数据量在 128MB 到 256MB 之间。如果你的输入是 20GB 的 Parquet 文件初始分区数量大概在 80 到 160 之间比较合适。这个只是经验值不同资源环境下要做微调。通过repartition(n)可以把分区数量改成你指定的值。读取数据时也可以通过设置spark.sql.files.maxPartitionBytes控制分区大小比如在读取超大文件时调低这个值让分区更细粒度有利于并行度提升。但要注意过小的分区会让元数据开销变大不一定更快。6. 常见报错与排查技巧实录照着操作就能省时间6.1 Task not serializable别再靠加 transient 硬扛了这个报错在 Scala 开发中极其常见。我曾经在某公司的调度平台上跑定时作业第一次出现这个报错时排查日志花了大半天。后来经验丰富了几十秒就能定位。起因是算子函数中引用了外部实例的方法或字段。比如下面这种class DataCleaner { val separator , def clean(rdd: RDD[String]) rdd.map(_.split(separator)(0)) }separator字段属于DataCleaner实例而DataCleaner这个类本身可能没有实现Serializable接口。当map里的函数被分发时JVM 试图序列化这个实例就报了 not serializable。解决方案有几种让DataCleaner继承Serializable。把separator提取为局部 val让 lambda 只捕获字符串而不是整个类实例。把方法体改成静态工具函数。我强烈推荐第二种因为第一种如果类里还有其他不可序列化成员照样会炸。局部变量是最干净的不会引入额外依赖。6.2 OutOfMemory到底应该加内存还是改代码Spark 作业报 OOM 时很多人的第一反应是调高 Executor 内存。这个可以但不总是最好的方向。如果是单个 Executor 内处理的数据量过大加内存只能暂时缓解真正要处理的是分区粒度和 shuffle 数据量。常见的优化路径用mapPartitions替代map前先想清楚每个分区会不会产生大量中间对象。检查是否在同一作业里对超大 DataFrame 重复执行了多个行动操作而没有任何缓存或复用。重建一个更大的模型用 broadcast 变量替代 join。减少不必要的collect操作。很多新手在调试时喜欢.show()和.collect()这两个操作会把大量数据拉回 DriverDriver 内存一爆就是 OOM。真要看数据用.take(10)别动不动就collect。如果确认是分区内数据实在过大那就用repartition增加分区数让每个分区处理的数据变小。6.3 找不到 Encoder 或类型不匹配Dataset 开发的两个高频编译错误使用 Dataset 写代码时最容易碰到的编译错误有Unable to find encoder for type stored in a Dataset意思是 Spark 不知道怎么把自定义类型序列化。解决办法是导入spark.implicits._如果还不行手动定义一个 Encoder。type mismatch则多是因为把字符串和 Column 混淆了。比如filter(age 18)这里的字符串会被当作 SQL 表达式解释而filter($age 18)是列表达式。两者效果类似但如果字符串拼错运行时报错会比编译期更隐蔽。我的习惯是能用 Column 表达式就用 Column 表达式。因为字符串表达式在编译时不检查列名一旦列名改了没同步改字符串运行时会抛出AnalysisException。Column 写法里用了$列名编译器不会保护字面量列名但至少代码结构更明确。当然最稳妥的还是用 Dataset 的强类型 filter编译期就发现字段不存在。6.4 小文件问题写数据的语法也要讲究用 Spark 产出数据时一个很常见的现象是生成了大量小文件。默认情况下Spark 写入操作的结果文件数量和最后的分区数量一致。如果你的 DataFrame 有几百个分区每个分区数据量又不大写出去就会产生几百个小文件对下游的读取非常不友好。处理办法很简单写前用coalesce或repartition调整分区数。coalesce更适合在需要减小分区数时使用因为它尽量不触发 shufflerepartition会重新均匀分布数据。要合并成少量大文件用repartition(8)或者coalesce(8)都行。但如果文件数量不是问题反而要小心过度合并导致单个文件过大。比如下游系统加载文件时内存有限文件太大就会加载失败。所以合并到什么程度需要根据下游需求评估。6.5 通过 explain 查看执行计划语法问题的最终裁判有时同一个需求能用好几种写法实现比如df.filter($a 1).select($b)和df.select($b).filter($a 1)最终的计算结果一样。但 Spark 优化器可能把它们转换成完全不同的物理执行计划。判断哪种写法效率更高的最终办法不是猜是explain()。df.filter($age 18) .groupBy($city) .agg(count(*)) .explain(true)explain(true)打印的是完整解析逻辑未解析的逻辑计划、解析后的逻辑计划、优化后的逻辑计划和物理计划。日常开发不用看那么细主要看Physical Plan里有没有Exchange表示发生了 shuffle、有没有BroadcastExchange、有没有Scan parquet之后马上就做过滤。如果发现了不必要的Exchange就要回头调整写法或加提示。我自己有个习惯写完一个稍复杂的作业在提交前用explain扫一眼物理计划看到广播 join 和合理的过滤顺序心里就踏实。这个习惯帮我挡掉了不少潜在的慢任务。我的几条实践经验Spark 里写 Scala 语法说到底是两条线一条是集合操作思维一条是分布式执行思维。前者靠多练 List、Seq 上的函数式操作后者靠理解算子什么时候触发 shuffle、什么时候可以本地执行。两条线交叉在一起就是开发 Spark 作业的主心骨。再补一个小技巧开发阶段可以用local[*]模式快速验证算子逻辑不用非得到远程集群上跑。每次改动后先在本地用小数据集跑通语法再提交调度平台做全量运行。这样能节约大量排队等待时间。最后说说怎么继续深入学习。最有价值的资料不是各种语法速查表而是 Spark 源码里那些直接用 Scala 写的算子和执行器代码。打开源码翻一翻看groupByKey是怎么实现的看mapPartitions的函数签名到底长什么样比死记硬背 API 深刻得多。接下来接手的项目不一定是新项目但只要你手头的数据处理任务里能用上这些语法就已经走在正确的路上了。
