企业官网建设流程全解析

1. 选型背后的逻辑：为什么要用Scala写Spark

1.1 先想清楚场景再选型

我最早接触Spark的时候用的是Python，因为上手快、写起来顺手。但等真正把任务推到几十台机器的集群上，遇到性能瓶颈和内存问题的时候，才发现Python那套在其他语言里能用的优化手段在Spark里经常使不上劲。后来有一次要接手一个别人用Scala写的流处理任务，我硬着头皮去读代码，读完之后才意识到，这个组合之所以成为工业界的主流搭配，不是没有道理的。

先说Spark本身解决的是什么问题。在单机时代，数据量再大也就是几GB、几十GB，用Pandas、用R都能扛。但数据到了TB甚至PB这个量级，单机的内存和CPU就完全不够用了。Spark的核心思路是：把数据拆成小块，分发到多台机器上并行处理，处理完再汇总。它靠的是两个东西——内存计算和DAG执行引擎。和MapReduce相比，Spark把中间结果尽量留在内存里，不用反复落盘，所以迭代式的计算（比如机器学习算法、图计算）能快出好几倍。

再说为什么是Scala。Spark本身就是用Scala写的，这就意味着用Scala写Spark程序，你能拿到几个实打实的好处。第一，API最全，任何新特性都是先出Scala版本，其他语言都得等；第二，性能好，Scala跑在JVM上，和Spark的底层执行引擎是同一个运行时，没有跨语言调用的开销；第三，类型安全，编译期就能发现一部分错误，不像Python要跑到一半才炸出来。加上Scala的函数式风格和Spark的算子模型几乎是无缝衔接，写起来行云流水。

当然，不是所有场景都非要Scala不可。如果只是做数据分析、写写即席查询，PySpark完全够用。但如果是要做严肃的、需要长期维护和反复调优的生产级数据处理应用，Scala是更稳的选择。我这篇实战记录，就是从一个完整的日志分析任务出发，把从零搭建到调优踩坑的全过程梳理一遍。

1.2 这套组合的适用边界

在决定用Spark+Scala之前，先把你的场景摸清楚。我见过不少团队一上来就上集群，结果数据量只有几百MB，纯属杀鸡用牛刀。一般来说，满足下面几个条件里的至少两个，才值得上Spark：

• 数据量在单机内存装不下的量级（大几十GB往上）；
• 计算逻辑里有迭代、多轮Join、复杂聚合这类需要中间结果复用的场景；
• 需要从批处理平滑扩展到流处理的长期规划；
• 对计算任务的稳定性和可维护性有明确要求。

反过来说，如果你的数据就是几GB的CSV，计算逻辑也只是过滤加分组，那老老实实用单机工具反而更快，不需要引入分布式带来的运维复杂度。选型和写代码一样，很多时候做减法比做加法更难。

2. 核心概念与Scala映射：把这五个关键点吃透

2.1 RDD、DataFrame、Dataset到底该用哪个

这是每个Spark新手都会问的问题，也是我在实际项目里被问得最多的问题。简单说，RDD是最底层的抽象，DataFrame是带Schema的分布式表，Dataset是类型安全的DataFrame。

用生活化类比来解释：RDD就像一堆散在地上的乐高积木，你想怎么拼都可以，但每次拼都要自己检查积木的形状对不对；DataFrame像一张Excel表，每列有明确的表头和数据格式，Spark可以根据列名自动做优化；Dataset则是给Excel表加了一层合同——每个单元格的类型在编译期就定死了，写错了代码直接编译不过，不用等运行到那一行才报错。

实际开发里，我的建议是这样的：

• 新项目一律从DataFrame或Dataset开始，除非你要做的底层自定义操作绕不开RDD；
• 对关键的数据模型用case class定义，然后转Dataset，享受编译期类型检查；
• 只在某些DataFrame API覆盖不到的底层场景（比如要自定义分区器、操作Partition内部数据）才动RDD。

原因很简单：DataFrame通过Catalyst优化器做执行计划优化，同一条逻辑，可能比你手写RDD操作快几倍。用生活类比就是，你自己一条条走迷宫（RDD），还不如让系统给你直接规划一条最优路（DataFrame）。

2.2 惰性求值和DAG执行引擎：别被表象骗了

Spark最反直觉的一点是：你写完一大堆转换操作，它其实什么都没干。只有遇到行动操作（Action）——比如count、collect、save——Spark才会真正去执行。这就是惰性求值。

这个设计的好处是，Spark可以在真正执行之前，把整条操作链看一遍，做各种优化：合并相邻的过滤条件、调整Join顺序、下推谓词等等。就像你出门前先规划好路线，而不是每条路都走一遍才发现错了。

但这也带来了一个调试上的坑：很多新手在DataFrame上写错了逻辑，打印出来的中间结果却是对的，因为Spark只执行了前面属于惰性的部分。等你加上行动操作想看一眼最终结果，才突然报错。所以我的习惯是，在关键节点上主动加一个cache()再配合count()去触发执行，既验证逻辑，又为后续复用做准备。

DAG（有向无环图）是Spark执行引擎的骨架。每个行动操作会触发一个Job，Job被拆成多个Stage，Stage之间按照依赖关系构成一个DAG。Stage内部的并行任务叫Task，每个Task处理一个分区的数据。理解这条链路，对后面的性能调优至关重要——因为很多问题的根源，都能在DAG的Stage划分里看出来。

2.3 case class和Encoder：类型安全的基石

Scala在Spark里最出彩的设计就是case class。它本质上是一个不可变的、自带equals/hashCode/toString的类，还支持模式匹配。在Spark里，case class最大的价值是可以作为Dataset的元素类型，Spark通过Encoder把它映射成内部二进制格式，绕开了Java/Scala序列化的开销。

我第一次意识到这个差别，是在处理一个几亿行的用户行为日志时。用RDD+自定义类的形式做聚合，跑了十几分钟没跑完；换成case class+Dataset，同样的逻辑只用了几分钟。原因就在于Spark对Dataset内部数据使用自己的二进制编码，操作的时候直接在二进制数据上做计算，不需要反复把对象序列化再反序列化。

定义一个case class很简单：

case class UserAction( userId: String, actionType: String, pageId: String, timestamp: Long, durationMs: Long )

然后就能在Dataset里做类型安全的操作了：

import spark.implicits._ val actionsDs: Dataset[UserAction] = spark.read .option("header", "true") .csv("/data/user_actions.csv") .as[UserAction]

这里有个细节要注意：case class的字段名要和表头对得上，字段类型也得匹配。timestamp如果CSV里是字符串，你需要先转成Long再映射，否则编译能过，运行时会出转换错误。

2.4 转换操作与行动操作的取舍

Scala集合API和Spark RDD/DataFrame的API非常像，map、filter、flatMap、groupBy、reduceByKey这些名字都一样，但背后的语义完全不同。Scala集合是立即求值的，Spark是惰性求值的；Scala集合操作的是内存里的对象，Spark操作的是分布在多台机器上的分区数据。

有一个常见的误区是用Spark的groupBy+reduce去做聚合，而不用reduceByKey。两者的差别在于数据混洗（Shuffle）的量：groupBy会把所有相同Key的数据全部拉到一起再处理，reduceByKey会在每个分区内先做一次本地聚合，再把聚合后的结果拉过去合并。在数据量大的时候，这个差别是数量级的。

另一个取舍是尽量少用collect()。在分布式环境里，collect()会把所有结果拉到Driver端，一旦数据量大，直接内存溢出。我通常只在调试小数据时用它，生产代码里宁可用save写到分布式存储，也不要冒这个险。

2.5 闭包序列化：新手最容易踩的无形陷阱

Scala写Spark时，很多代码看起来没问题，一提交就报Task not serializable。这是Spark里最经典的问题，前排提示一下根因：Spark会把你在算子里写的闭包（lambda表达式）发给所有Executor节点，闭包里引用的外部对象必须能序列化，否则就炸。

解决办法有这么几条，从适用面宽到窄排列：

• 把大对象（比如某个配置类）定义成@transient lazy val，让它不参与序列化；
• 把广播变量用Broadcast包装，避免每个Task都复制一份大对象的副本；
• 在驱动器里把需要的字段提取成局部变量，闭包里只引用这个局部变量，不要引用整个对象。

我比较推荐第三个办法，因为它最简单直接，也符合函数式编程的风格：闭包只捕获它需要的最小上下文。

3. 从零搭建一个实战项目：用户行为日志分析系统完整落地

3.1 项目初始化：SBT配置与目录结构

我用的是SBT作为构建工具，因为它对Scala项目的支持最成熟，几乎就是Scala世界的Maven。以下是一个标准配置的build.sbt，可以直接抄作业：

ThisBuild / scalaVersion := "2.12.18" ThisBuild / organization := "com.example" ThisBuild / version := "1.0.0" lazy val root = (project in file(".")) .settings( name := "log-analysis", libraryDependencies ++= Seq( "org.apache.spark" %% "spark-core" % "3.5.1" % "provided", "org.apache.spark" %% "spark-sql" % "3.5.1" % "provided", "org.scalatest" %% "scalatest" % "3.2.17" % Test ) )

注意两点：Spark相关依赖用了provided作用域，意味着编译和测试时用，打包时不打进去，因为目标集群上已经装了Spark。这样打出来的Jar体积小很多。Scala版本要用%%选，它会自动匹配和当前Scala版本一致的人工产物。

目录结构我一般这样组织：

src/main/scala/ ├── JobMain.scala // 入口，负责参数解析和任务调度 ├── pipeline/ │ ├── DataLoader.scala // 数据加载与清洗 │ ├── MetricsCalculator.scala // 核心计算逻辑 │ └── ResultWriter.scala // 结果输出 └── model/ └── UserAction.scala // 数据模型（case class）

把入口、数据处理、计算逻辑、模型分开，好处是后续要换数据源或者改计算逻辑，不用整个重读一遍代码。

3.2 数据模型与数据加载：先定Schema再谈性能

假设我们要处理的是埋点日志，每条记录包含用户ID、行为类型、页面ID、时间戳、停留时长。先定义case class：

case class UserAction( userId: String, actionType: String, pageId: String, eventTime: Long, durationMs: Long )

然后初始化SparkSession：

val spark = SparkSession.builder() .appName("UserActionLogAnalysis") .config("spark.sql.shuffle.partitions", "200") .config("spark.executor.memory", "4g") .getOrCreate() import spark.implicits._

读取原始日志时，我的经验是不要直接让Spark去推断Schema，而是先在地端做一次格式探查，把字段类型定死，再往Spark里灌。因为Spark的Schema推断在数据量大时要额外扫一遍全量数据，成本很高；而且自动推断的结果经常是String类型偏多，后面还要再Cast一次，纯属浪费时间。

val rawDf: DataFrame = spark.read .option("header", "true") .option("inferSchema", "false") .csv("/data/raw_logs/2025/")

然后手动映射成Dataset：

val actionsDs: Dataset[UserAction] = rawDf .select( $"user_id".cast(StringType).as("userId"), $"action_type".cast(StringType).as("actionType"), $"page_id".cast(StringType).as("pageId"), $"event_time".cast(LongType).as("eventTime"), $"duration_ms".cast(LongType).as("durationMs") ) .as[UserAction]

这一步看似繁琐，但它带来了两个好处：后续所有的计算都在强类型环境下进行，编译期能发现的错误不会拖到运行时；Spark对Dataset内部存储的二进制格式读取效率远高于Row对象，实际跑下来能差出不少时间。

3.3 核心计算逻辑：从过滤、聚合到窗口函数

现在开始做正事。需求是这样的：按天统计每个页面的PV、UV、平均停留时长，以及每个用户一天内的访问路径（按时间排序的页面序列）。前两个用DataFrame API就够了，第三个需要窗口函数。

先做最基础的清洗：过滤掉字段缺失或者时间戳为0的脏数据：

val cleaned = actionsDs .filter(_.userId.nonEmpty) .filter(_.eventTime > 0L) .filter(_.durationMs >= 0L)

然后计算PV：

val pvResult = cleaned .groupBy($"pageId") .count() .withColumnRenamed("count", "pv")

计算UV需要对userId去重：

val uvResult = cleaned .groupBy($"pageId") .agg(countDistinct($"userId").as("uv"))

PV和UV的一个关键区别：PV是所有人点了多少次，UV是点过的人有多少个。在大数据量下，countDistinct的开销不小，因为它要做全局去重。如果只是估算UV，可以用approx_count_distinct（基于HyperLogLog算法），误差在1%以内，但速度能快一个量级。

平均停留时长就是聚合加平均：

val avgDuration = cleaned .groupBy($"pageId") .agg(avg($"durationMs").as("avgDurationMs"))

把三个结果合并成一个宽表可以这样做，注意这里我用了一个隐式转换的写法：

val dailyStats = pvResult .join(uvResult, Seq("pageId"), "left_outer") .join(avgDuration, Seq("pageId"), "left_outer")

然后是计算访问路径。访问路径本质上是"同一用户在同一时间段内访问的页面序列"，这里我先用一个简单的方式——按用户分组，时间排序后收集列表：

import org.apache.spark.sql.expressions.Window val pathWindow = Window.partitionBy($"userId").orderBy($"eventTime") val paths = cleaned .withColumn("rn", row_number().over(pathWindow)) .groupBy($"userId") .agg(collect_list($"pageId").as("pagePath"))

需要注意，collect_list的顺序并不是天然有保证的，所以上面的代码里我先加了row_number，确保取出来的是排序后的页面序列。这是一个我在实际项目里踩过坑的细节：不排序直接collect，拿到的路径经常是乱序的。

3.4 性能优化手段：分区、缓存与广播变量

代码写对了只是第一步，能不能在合理时间内跑完是另一回事。这里分享我在这个项目里实际用到的三个优化手段。

第一是合理设置分区数。分区数决定了并行度——分区太少，CPU资源闲置；分区太多，调度开销和Shuffle成本飙升。我的经验是：目标分区数=集群总核心数×2~3。比如10台机器、每台4核，那就是40个核心，分区数设在80~120之间比较合适。可以通过repartition调整：

val balanced = cleaned.repartition(96, $"userId")

用userId做分区键还有一个额外的好处：同一个用户的所有行为落到同一个分区，后续涉及用户的聚合操作可以避免跨分区的Shuffle。

第二是巧用缓存。如果一个DataFrame会被后续多个操作反复读取，缓存它一次能省下大量重复计算。这里有个细节：cache()是惰性的，必须跟一个行动操作触发它。我通常这样写：

val cached = cleaned.cache() cached.count() // 触发缓存

缓存的存储级别默认是MEMORY_ONLY，但实际开发中我更常用MEMORY_AND_DISK——内存放不下就落盘，至少不会因为OOM把整个任务搞挂。在配置里预留：

.config("spark.rdd.compress", "true") .config("spark.sql.inMemoryColumnarStorage.compressed", "true")

第三是广播变量。当一个小表（比如维表、配置表）要被每个Task使用时，把它广播出去，避免每个Task都从Driver拉一份完整副本。比如我们要把页面名称映射表广播出去：

val pageNameMap: Map[String, String] = loadPageNameMap() val pageNameBc = spark.sparkContext.broadcast(pageNameMap) val enriched = cleaned.map { action => val pageName = pageNameBc.value.getOrElse(action.pageId, "unknown") (action.pageId, pageName, action.durationMs) }

广播变量的大小一般建议小于100MB，超过这个量级广播本身的网络开销就会反噬性能。用生活类比就是：让每个人手里都拿一份地图（穿版），比每个人每次都跑回总部问路要快得多。

3.5 结果输出：写出分区表和检查点机制

最后把计算结果写入分布式存储。写入时有一个关键问题：小文件过多。如果输出数据的每个分区只有几百KB，会产生成千上万个碎片文件，后续查询这些文件时，打开文件本身的开销可能远超读取内容的时间。

解决小文件问题有两个常用手段。一是先coalesce或repartition到合适的输出分区数：

stats.coalesce(24).write.mode("overwrite").parquet("/data/stats/daily")

二是开启Spark 3.0+的动态分区裁剪，让Spark自动合并小分区：

.config("spark.sql.adaptive.enabled", "true") .config("spark.sql.adaptive.coalescePartitions.enabled", "true")

对于流式计算场景，我强烈建议在关键检查点加上Checkpoint机制。它是容错的关键：Spark定期把计算状态存到可靠的分布式存储上，任务失败后能恢复进度，而不用从头算起。这里我用的是结构化流式的写法，先不展开，后面单开一节细说。

4. 我踩过的那些坑：Spark+Scala常见问题排查实录

4.1 数据倾斜：任务卡在99%的真相

数据倾斜是整个Spark开发里最常见的性能杀手。现象很典型：跑一个任务，99%的任务早就完成了，就剩一两个任务拖着整个Stage，跑几个小时都不结束。原因也很简单：某个Key的数据量远超其他Key，导致那一个分区的计算量极大。

我遇到过一次真实案例：用户行为日志里，有一个测试用户的访问量是正常用户的上万倍。所有聚合都卡在那一个Key上。

解决方法通常是加盐（Salting）。做法是给倾斜的Key加上随机前缀，把一个大Key拆成多个小Key：

// 先定位倾斜Key val skewedKey = "test_user_001" val salted = cleaned.map { action => if (action.userId == skewedKey) { val salt = scala.util.Random.nextInt(100) (s"$skewedKey#$salt", action.durationMs) } else { (action.userId, action.durationMs) } }.groupByKey(_._1) .agg(avg("_2")) // 对加盐后的key分别聚合

但加盐是有代价的。加盐之后，同一个Key的数据被拆开了，聚合结果要再做一层合并。所以这个方案只针对明确识别出的少数倾斜Key，不要全表加盐，那会引入大量不必要的Shuffle。

另一个办法是广播Join。如果小表本身不大，直接用Broadcast Join，让Spark把小表复制到每个Executor，完全跳过Shuffle阶段。用生活类比就是：与其让几千个人排队去一个窗口办事，不如在每个窗口都放一份办事指南，大家就地解决。

4.2 Executor OOM：内存设置的排列组合

内存溢出在Spark里分两种：Executor端OOM和Driver端OOM。大部分情况下是前者。

我总结过一个排查顺序：

1. 看是不是数据倾斜导致的单分区内存爆炸（按上面的方法处理）；
2. 看是不是Shuffle时spark.sql.shuffle.partitions设置得太少，导致每个分区承接的数据过多；
3. 看缓存是否设得过大；Spark的堆内内存和堆外内存要分开配置，spark.executor.memory是堆内，spark.memory.offHeap.enabled和spark.memory.offHeap.size控制堆外，堆外能减少GC压力，但需要谨慎分配；
4. 如果是Driver端OOM，大概率是collect()了太多数据，检查逻辑里有没有不小心把大结果拉到Driver。

我推荐的一个默认配置模板（以单Executor 8GB内存为例）：

.config("spark.executor.memory", "8g") .config("spark.executor.memoryOverhead", "2g") .config("spark.memory.fraction", "0.8") .config("spark.memory.storageFraction", "0.5")

memoryOverhead是给JVM之外的本地内存用的，比如线程栈、NIO buffer，千万别设成0。storageFraction是执行内存和存储内存的分界线，如果你的任务缓存用得多，可以把存储比例调高；如果计算密集、Shuffle多，把执行内存调高。

4.3 Task not serializable：闭包里不要放不该放的东西

这个报错我估计每个写Spark的人都会遇到一遍。报错的完整信息一般是Task not serializable或者org.apache.spark.SparkException: Task not serializable。触发原因我在前面2.5节提过：闭包捕获了不能序列化的对象。

举个典型的错误代码：

// 错误示例 class MetricsCalculator { val regex = new Regex("...") def process(ds: Dataset[UserAction]): DataFrame = { ds.map { action => val matched = regex.findFirstIn(action.pageId) // 这里捕获了regex ... }.toDF() } }

这段代码里，regex属于MetricsCalculator实例，闭包里引用了它，导致整个MetricsCalculator实例被要求序列化。而如果MetricsCalculator里还有数据库连接或者文件句柄，那就不可能序列化成功。

正确的改法是：把正则表达式提取成局部变量：

def process(ds: Dataset[UserAction]): DataFrame = { val regex = new Regex("...") // 局部变量 ds.map { action => val matched = regex.findFirstIn(action.pageId) ... }.toDF() }

闭包只捕获了regex这个局部变量，它本身可以序列化，问题就解决了。

4.4 数据倾斜、Shuffle和分区数的联动排查

最后说一个排查经验：数据倾斜、Shuffle量、分区数这三个问题经常连环出现。

举例说明。有一次我们的任务跑得很慢，我在Spark UI上看Shuffle Read的量惊人了——居然有几十GB。排查后发现，根因是前面某个算子里两个DataFrame做Join时，Join Key的分布严重不均，导致Shuffle阶段出现数据倾斜；再往上一层看，发现分区数设置得过小，只有50个分区，而集群有80个核心，大量CPU直接闲置。

三个问题其实是同一个链路：分区不合理引起了Shuffle放大，Shuffle放大引发了倾斜，倾斜拖慢了整体进度。所以调优时不要孤立地看单个指标，要连起来看。我现在排查性能问题时的习惯是：

• 先在Spark UI上看每个Stage的Shuffle Read/Write量，找出异常的Stage；
• 再进那个Stage看Task的执行时间分布，如果少数Task时间和多数Task差太多，大概率是倾斜；
• 最后根据核心数调整分区数，一般让分区数保持为核心数的2~3倍。

用生活类比来说，这就像诊断交通拥堵：不能只看某一路口的车流量，还要看整条路网的通行能力、红绿灯配时和高峰期的分布，三件事联动着看才能找到真正的堵点。

5. 一些我在实际项目中反复用的调优清单

5.1 编码和配置习惯

项目里采用下面这些配置后，整体稳定性提升明显。这些参数不是万能药，但作为起步配置很稳：

.config("spark.sql.adaptive.enabled", "true") .config("spark.sql.adaptive.coalescePartitions.enabled", "true") .config("spark.sql.adaptive.skewJoin.enabled", "true") .config("spark.sql.shuffle.partitions", "200") .config("spark.sql.files.maxPartitionBytes", "134217728") // 128MB

前三个是Spark 3.0之后引入的Adaptive Query Execution，它能根据运行时统计动态调整执行计划。skewJoin.enabled开起来后，Spark会自动把小表广播到倾斜Key所在的执行器，很多数据倾斜问题在不需要改代码的情况下就能缓解一大半。

我把spark.sql.files.maxPartitionBytes设为128MB，配合128MB的HDFS块大小，这样每个文件块对应一个分区，读取效率最高。如果设得太大，分区数变少，并行度下降；设得太小，分区数暴增，调度开销上来。128MB是一个经过大量实践验证的合理默认值。

5.2 用Scala特性把代码写得更优雅

函数式风格、模式匹配和隐式转换这三个Scala特性，是Spark代码的加分项。

函数式风格让代码天然贴合Spark算子。比如多分支处理用match比用一长串if else清晰太多：

action.actionType match { case "click" => handleClick(action) case "view" => handleView(action) case "share" => handleShare(action) case other => handleDefault(other) }

隐式转换在Spark里最常见的应用是DataFrame和Dataset之间的互相转换：导入spark.implicits._之后，Scala的集合就能直接变成DataFrame，case class也能变成Dataset。这个特性让代码简洁不少，但要注意，隐式转换很容易让团队新人读不懂，所以在我们团队里有一条不成文的规矩：隐式转换只用于项目内约定的几种场景，不随意外扩。

5.3 代码可读性比炫技重要

去年接手一个老项目，里面有一堆用了复杂隐式转换和泛型嵌套的代码。功能倒是都对的，但任何人来维护都叫苦不迭。我自己现在写代码有一个原则：代码的首要读者是人，其次才是机器。

具体到Spark+Scala，我认为可读性的优先级排序是这样的：

• 先写明白的业务逻辑，再考虑性能优化；
• 命名要表达意图，不要用df1、df2这种变量名；
• 超过三行以上的逻辑就应该拆成函数，加上注释；
• 性能优化加注释说明"为什么这么做"，不然三个月后你自己也会忘记当初为什么要加这个盐。

这段经验，比任何具体的API用法都值得记住。

5.4 从批处理到流处理的无缝升级

使用Spark做批处理的项目，往往有朝流处理演进的需求。Scala写的批处理代码有一个天然优势：逻辑可以复用。

比如上面那个日志分析系统，批处理每天跑一次，计算全量一天的PV、UV。慢慢需求变成"想看实时的PV、UV"，只需要很小的改动就能把批处理代码迁移到Structured Streaming：

val streamInput = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "localhost:9092") .option("subscribe", "user_actions") .load() val parsed = streamInput .selectExpr("CAST(value AS STRING)") .select(from_json($"value", schema).as("data")) .select($"data.*") val query = parsed .groupBy($"pageId") .agg(count($"*").as("pv"), countDistinct($"userId").as("uv")) .writeStream .outputMode("update") .format("console") .start() query.awaitTermination()

核心的聚合逻辑几乎不用改，因为Structured Streaming用的也是DataSet/DataFrame API。这就是选型时选Scala的长期优势——批流一体，成本最低。

6. 最后想说的话

这篇实战记录里我尽量避开了那些教科书式的"默认操作"，把真正在项目里反复验证过的东西挑了出来。包括为什么选Scala、Dataset和DataFrame怎么取舍、闭包序列化怎么避免、数据倾斜怎么加盐、调优时怎么联动看指标。这些内容单独每一条都算不上新奇，但组合在一起，确实能让一个刚入门的开发者少走很多弯路。

我最想强调的还是那句话：写Spark代码，逻辑正确只是起点，跑得动、跑得快、好维护才是终点。而这三个目标，恰好是Scala这个语言和Spark这个框架组合在一起时最擅长达成的事情。如果看完这篇，你能少踩一个坑，或者踩坑之后能更快找到解法，那这篇就算没白写。