1. 从零开始：搭建你的第一个Spark实战环境

很多朋友一听到Spark，就觉得是那种只有大公司才用得上的“重型武器”，心里先打起了退堂鼓。其实不然，Spark的入门门槛远没有想象中那么高。我自己刚开始接触的时候，也是从一台普通的笔记本电脑开始的。今天，我就带你一步步搭建一个能跑起来的Spark环境，咱们不搞那些虚的，直接上手操作，目标是让你在半小时内，写出并运行第一个Spark程序。

首先，咱们得把“地基”打好。Spark运行需要Java环境，我推荐安装OpenJDK 8或者11，这两个版本和目前主流的Spark 3.x兼容性最好。在Ubuntu系统里，打开终端，一行命令就能搞定：sudo apt install openjdk-11-jdk-headless。装好后，用 java -version 确认一下，能看到版本号就说明成功了。

接下来是主角Spark。我强烈建议直接从Apache官网下载预编译好的版本，省去自己编译的麻烦。比如，我们可以下载Spark 3.3.2 with Hadoop 3.3。下载完是个 .tgz 压缩包，把它解压到你习惯的目录，比如 /home/你的用户名/spark。然后，需要配置几个环境变量，让系统知道Spark在哪。编辑你的 ~/.bashrc 文件，在末尾加上这么几行：

export SPARK_HOME=/home/你的用户名/spark
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
export PYSPARK_PYTHON=python3

保存后，执行 source ~/.bashrc 让配置生效。现在，在终端里输入 spark-shell 并回车，如果你看到那个熟悉的Spark Logo和Scala交互式命令行提示符 scala> 蹦出来，恭喜你，Spark单机环境已经成功启动了！这个 spark-shell 是我们后续做快速测试和探索的利器，就像Python的IDLE或者Jupyter Notebook一样方便。

不过，光有Spark还不够，我们处理的数据往往来自文件。这里有两种主要的文件系统需要了解：本地文件系统和HDFS。本地文件系统就是你自己电脑的硬盘，路径像 /home/hadoop/test.txt 这样。而HDFS全称是Hadoop分布式文件系统，它是为存储海量数据而设计的，路径通常像 hdfs://localhost:9000/user/hadoop/test.txt。对于初学者，我建议先从本地文件开始玩起，等熟悉了基本操作，再尝试连接HDFS。如果你想在单机上模拟HDFS，可以安装Hadoop单节点模式，但这会稍微复杂一点。咱们今天的实战，为了求快求稳，先从本地文件开始。

2. Spark编程第一课：与文件系统对话

环境准备好了，手就开始痒了，对吧？咱们立刻开始写代码。Spark的核心抽象叫做弹性分布式数据集，简称RDD，以及它的高级版本DataFrame/Dataset。刚开始，咱们先用最基础的RDD来感受一下Spark的编程模型，它非常直观。

2.1 在spark-shell里快速体验文件读取

打开你的spark-shell。首先，我们试试读取本地的一个文本文件。假设我在 /home/hadoop 目录下已经准备了一个 test.txt 文件，里面随便写了几行文字。在spark-shell里，输入以下代码：

val localFile = sc.textFile("file:///home/hadoop/test.txt")
val lineCount = localFile.count()
println(s"本地文件的行数是: $lineCount")

看，是不是很简单？sc 是SparkContext，它是Spark所有功能的入口，在spark-shell里已经自动为你创建好了。textFile 方法用来读取文本文件，它返回一个RDD，里面的每个元素就是文件的一行。count() 是一个行动操作，它会触发真正的计算，数一数RDD里有多少个元素，也就是文件有多少行。

那么读取HDFS上的文件呢？假设你的HDFS服务已经启动，并且文件上传到了 /user/hadoop/test.txt。代码几乎一样，只是路径前缀换了：

val hdfsFile = sc.textFile("hdfs://localhost:9000/user/hadoop/test.txt")
val hdfsLineCount = hdfsFile.count()
println(s"HDFS文件的行数是: $hdfsLineCount")

这里的关键是协议头：file:// 表示本地文件，hdfs:// 表示HDFS文件。我刚开始学的时候，经常忘记加 file://，导致Spark去HDFS上找文件，当然就找不到了，报错能让人排查半天。所以这是一个要记住的小坑。

2.2 迈出独立应用的第一步：打包与提交

在shell里玩转之后，我们肯定要写独立的应用程序。这才是真正的项目开发方式。咱们用经典的“统计文件行数”这个任务来走通全流程。我用Scala语言来写，因为它和Spark是天作之合，表达起来非常简洁。

首先，创建一个项目目录，比如 SimpleApp。在里面创建两个关键文件。第一个是源代码文件 src/main/scala/SimpleApp.scala：

import org.apache.spark.{SparkConf, SparkContext}

object SimpleApp {
  def main(args: Array[String]): Unit = {
    // 1. 创建Spark配置，设置应用名称
    val conf = new SparkConf().setAppName("Simple Application")
    // 2. 创建SparkContext，它是所有功能的入口
    val sc = new SparkContext(conf)

    // 3. 读取HDFS上的文件。这里路径可以通过args参数传入，更灵活。
    val logFile = "hdfs://localhost:9000/user/hadoop/test.txt"
    val logData = sc.textFile(logFile, 2) // 第二个参数是最小分区数，可选

    // 4. 执行行动操作：计数
    val numLines = logData.count()

    // 5. 打印结果
    println(s"文件 $logFile 一共有 $numLines 行。")

    // 6. 停止SparkContext，释放资源
    sc.stop()
  }
}

第二个是项目构建文件 simple.sbt（放在项目根目录）：

name := "Simple Project"
version := "1.0"
scalaVersion := "2.12.17" // 注意版本要与Spark兼容
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.3.2"

这里我用了Scala 2.12和Spark 3.3.2，你需要根据自己下载的Spark版本调整。接下来，在项目根目录打开终端，运行 sbt package。sbt会自动下载依赖，并把你的应用打包成一个JAR文件，通常位于 target/scala-2.12/simple-project_2.12-1.0.jar。

最后，也是最激动人心的一步，用 spark-submit 提交任务：

$SPARK_HOME/bin/spark-submit \
  --class SimpleApp \
  --master local[2] \
  /path/to/your/project/target/scala-2.12/simple-project_2.12-1.0.jar

--master local[2] 表示在本地运行，并使用2个CPU核心。当你看到终端输出“文件...一共有...行”时，你的第一个独立Spark应用就成功运行了！这个过程看似步骤多，但只要你成功跑通一次，以后就是固定的流程，非常机械化。我建议你反复练习几次，直到不用看笔记也能完成。

3. 进阶实战：用Spark解决真实的数据清洗问题

掌握了基础的读写和提交，我们就可以挑战更实际的问题了。数据处理中，重复数据和无意义数据（脏数据）就像米饭里的沙子，必须剔除。下面我们通过两个实战案例，来学习Spark如何高效地进行数据清洗和转换。

3.1 案例一：多文件合并与精准去重

假设你是某电商平台的数据工程师，每天会收到来自不同数据源的用户行为日志文件A和B。它们可能有重复记录，你的任务是把它们合并，并去除所有重复的行，生成一份干净的汇总文件C。原始数据可能很大，用Excel根本打不开，但用Spark就能轻松处理。

我们模拟一下数据。文件A和B的内容就是任务描述里的样子，每一行是一条记录。思路很清晰：读取两个文件 -> 合并成一个RDD -> 去重 -> 保存。但这里有个关键点：什么是“重复”？ 在这个例子里，整行内容完全一样才算重复。比如“20170101 x”和另一个“20170101 x”是重复的，但“20170101 x”和“20170101 y”不算。

基于这个逻辑，我们编写独立应用 RemDup.scala：

import org.apache.spark.{SparkConf, SparkContext}

object RemDup {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Remove Duplicates")
    val sc = new SparkContext(conf)

    // 假设文件A和B的路径通过参数传入，这里我们用硬编码示例
    val fileAPath = "file:///home/hadoop/data/inputA.txt"
    val fileBPath = "file:///home/hadoop/data/inputB.txt"

    // 读取两个文件，得到两个RDD
    val rddA = sc.textFile(fileAPath)
    val rddB = sc.textFile(fileBPath)

    // 合并两个RDD
    val unionRDD = rddA.union(rddB)

    // 去除重复行：distinct() 是Spark RDD提供的去重方法，非常方便
    val distinctRDD = unionRDD.distinct()

    // 按原始需求，可能需要排序（根据日期）。先过滤掉空行，再排序。
    val cleanedAndSortedRDD = distinctRDD.filter(_.trim.nonEmpty).sortBy(identity)

    // 将结果保存到本地文件系统
    cleanedAndSortedRDD.saveAsTextFile("file:///home/hadoop/data/outputC")

    sc.stop()
    println("去重完成，结果已保存至 outputC 目录。")
  }
}

这里我用了 distinct() 方法，它是Spark RDD内置的去重算子，底层会经过shuffle过程，能保证全局去重。sortBy(identity) 表示按照元素本身（即整行字符串）进行排序。保存结果时要注意，saveAsTextFile 会生成一个目录，里面包含多个分区文件（part-00000等），这是分布式存储的特性。如果你想得到一个单一文件，可以在保存前使用 coalesce(1) 或 repartition(1) 将RDD合并成一个分区，但这样做会失去并行度，只适合小数据量结果。

3.2 案例二：多学科成绩统计与平均值计算

第二个场景更贴近业务：计算每个学生多门课程的平均分。数据来自三个文件，分别记录Algorithm、Database、Python的成绩。我们需要按学生姓名分组，然后计算其所有成绩的平均值。

这个任务比单纯去重更进一步，因为它涉及分组和聚合。我们分析一下步骤：1. 读取所有文件；2. 将每一行数据解析成 (学生姓名, 分数) 的键值对形式；3. 按学生姓名分组；4. 在每个分组内，计算分数的平均值。

来看代码实现 AvgScore.scala：

import org.apache.spark.{SparkConf, SparkContext}

object AvgScore {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Average Score Calculator")
    val sc = new SparkContext(conf)

    // 使用通配符一次性读取所有以‘score’开头的文件
    val dataFiles = "file:///home/hadoop/data/score*.txt"
    val linesRDD = sc.textFile(dataFiles)

    // 数据清洗与转换：过滤空行，并映射为 (name, score) 对
    val nameScorePairs = linesRDD
      .filter(_.trim.nonEmpty) // 过滤空行
      .map { line =>
        val fields = line.split("\\s+") // 按空白字符分割，更健壮
        (fields(0), fields(1).toDouble) // 转换为(姓名，分数)
      }

    // 核心操作：按姓名分组，并计算平均分
    val avgScoresRDD = nameScorePairs
      .groupByKey() // 按姓名分组，得到 (name, Iterable[score])
      .mapValues { scores => // 对每个学生的分数迭代器进行操作
        val scoreList = scores.toList
        val sum = scoreList.sum
        val count = scoreList.size
        // 计算平均分，并保留两位小数
        BigDecimal(sum / count).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
      }
      .sortByKey() // 按姓名排序，使输出更有序

    // 格式化输出为 (姓名, 平均分) 的形式并保存
    val resultRDD = avgScoresRDD.map { case (name, avg) => s"($name, $avg)" }
    resultRDD.saveAsTextFile("file:///home/hadoop/data/avg_score_result")

    sc.stop()
    println("平均分计算完成，结果已保存。")
  }
}

这段代码有几个值得注意的细节。第一，我用了 sc.textFile(dataFiles) 中的通配符 *，这样可以一次性读取所有符合模式的文件，非常方便。第二，在分割行数据时，用了 split("\\s+") 而不是 split(" ")，这样能处理多个空格或制表符的情况，代码更健壮。第三，也是最重要的，我们使用了 groupByKey() 后接 mapValues 的经典组合。groupByKey 会将所有相同键的值聚集到一起，形成一个可迭代的集合，然后在 mapValues 里我们对这个集合进行求和与计数。最后，我用 BigDecimal 来处理四舍五入，确保财务或成绩计算的精度。

4. 避坑指南与性能优化初探

跟着上面的例子做，你应该能成功运行程序了。但真实项目远比示例复杂，会遇到各种“坑”。这里我分享几个自己踩过、并且新手最容易遇到的坑，以及一些简单的优化思路。

第一个大坑：资源不足与OOM（内存溢出）。 如果你在本地用 local 模式跑，默认给Spark的内存可能很小。处理稍大的文件时，很容易就报 java.lang.OutOfMemoryError。解决方法是在 spark-submit 时指定更多的内存：

$SPARK_HOME/bin/spark-submit \
  --class YourApp \
  --master local[2] \
  --driver-memory 2g \ # 给Driver进程分配2GB内存
  --executor-memory 1g \ # 如果有Executor，也分配内存
  your-app.jar

第二个坑：数据倾斜。 这在 groupByKey、reduceByKey 等操作中特别常见。比如计算平均分的例子，如果有一个叫“张三”的学生有上百万条记录（可能是数据错误），而其他学生只有几条，那么处理“张三”的那个任务就会特别慢，拖垮整个作业。解决数据倾斜是个高级话题，但有一些基础手段：1. 考虑能否用 reduceByKey 或 aggregateByKey 在合并前先在本地进行聚合，减少shuffle数据量；2. 过滤掉异常多的脏数据；3. 对倾斜的Key进行加盐（Salt）处理，即给它加上随机前缀，打散后再聚合。

第三个坑：文件路径和序列化。 就像前面提到的，file:// 和 hdfs:// 要分清。另外，如果你在算子（如 map、filter）内部使用了一个外部定义的变量，这个变量需要能被序列化，否则会报 Task not serializable 错误。一个常见的例子是使用了某个类的成员变量。简单的解决办法是，将该变量标记为 @transient 或在算子内部重新定义。

关于性能的一点小建议： 在读取文件时，sc.textFile 的第二个参数可以指定最小分区数。分区数决定了并行度。如果文件很大，适当增加这个数字（比如设为CPU核心数的2-3倍）可能会提升读取速度。但也不是越大越好，分区太多会导致任务调度开销变大。对于 distinct() 和 groupByKey() 这类会引起shuffle的操作，结果RDD的分区数默认等于父RDD的分区数，你可以根据数据量使用 repartition() 或 coalesce() 进行调整。

最后，养成查看Spark Web UI的习惯。提交应用后，Spark会在4040端口（默认）提供一个Web界面，里面详细展示了作业的各个阶段（Stages）、任务（Tasks）、执行时间、数据 shuffle 量等信息。这是你分析应用性能、定位瓶颈的最直观工具。多看看，你就能慢慢理解你的代码到底是如何在集群上被拆分成任务并行执行的了。

写Spark程序就像搭积木，核心算子就那些，但组合起来能解决非常复杂的问题。从简单的文件行数统计，到数据清洗、聚合计算，再到更复杂的连接（Join）、窗口函数，都是一步步积累的。关键是多写、多跑、多遇到问题、多解决问题。我刚开始的时候，一个简单的任务因为数据倾斜卡了几个小时，查日志、调参数、改代码，最后解决的那一刻，那种成就感就是学习技术最大的乐趣。希望你能从这篇实战指南开始，享受用Spark处理数据的乐趣。如果在实践中遇到具体问题，不妨多看看官方文档和社区讨论，那里有海量的实战经验。