从两个CSV文件到HDFS结果：Spark求Top N支付额的保姆级避坑记录（含SBT安装报错解决）

当你第一次尝试用Spark处理真实数据时，可能会被各种环境配置和报错信息搞得焦头烂额。本文将带你完整走一遍从原始CSV文件到最终Top 5支付额结果的实战流程，重点解决那些教程里不会告诉你的"坑"。

1. 环境准备与数据上传

在开始编写Spark代码前，我们需要确保HDFS和Spark环境已经就绪。假设你已经安装好了Hadoop和Spark集群（单机伪分布式模式也适用），让我们从最基础的文件上传开始。

常见坑点1 ：HDFS路径错误是新手最容易犯的错误之一。很多人会直接复制教程中的 hdfs://ly1:9000 而不修改，导致后续程序无法找到文件。正确的做法是先检查你的HDFS配置：

# 查看HDFS的根目录地址
hdfs getconf -confKey fs.defaultFS

得到输出可能是 hdfs://localhost:9000 或 hdfs://your-hostname:9000 。记下这个地址，后续所有操作都要使用相同的地址。

接下来是文件上传的标准操作流程：

# 创建HDFS目录（注意替换your-hostname）
hadoop fs -mkdir -p hdfs://your-hostname:9000/example

# 上传文件（假设file1.txt和file2.txt在当前目录）
hadoop fs -put file1.txt file2.txt hdfs://your-hostname:9000/example

# 验证文件是否上传成功
hadoop fs -ls hdfs://your-hostname:9000/example

如果遇到权限问题，可以临时关闭HDFS权限检查（仅限开发环境）：

# 修改hdfs-site.xml添加以下配置
<property>
  <name>dfs.permissions.enabled</name>
  <value>false</value>
</property>

2. SBT安装与项目配置

Scala项目的构建工具SBT是另一个"坑"密集区。以下是经过验证的安装步骤：

1. 下载正确的版本 ：

   • 访问 sbt官网
   • 选择1.9.x系列版本（与Spark 3.x兼容性最好）

2. 安装过程 ：

# 创建安装目录
mkdir -p /opt/sbt

# 解压下载的压缩包
tar -zxvf sbt-1.9.9.tgz -C /opt/sbt

# 创建启动脚本
cat > /opt/sbt/sbt << 'EOF'
#!/bin/bash
SBT_OPTS="-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled"
java $SBT_OPTS -jar $(dirname $0)/sbt-launch.jar "$@"
EOF

# 赋予执行权限
chmod +x /opt/sbt/sbt

# 添加到PATH
echo 'export PATH=$PATH:/opt/sbt' >> ~/.bashrc
source ~/.bashrc

遇到"sbt-launch.jar找不到"怎么办？

这是最常见的SBT安装问题，解决方法：

# 进入sbt安装目录
cd /opt/sbt

# 确保jar文件存在
ls -l sbt-launch.jar

# 如果不存在，从bin目录复制
cp bin/sbt-launch.jar ./

# 验证安装
sbt sbtVersion

3. 创建Spark项目结构

标准的Scala项目目录结构如下：

sparkapp/
├── build.sbt
└── src/
    └── main/
        └── scala/
            └── TopN.scala

创建这个结构的命令：

mkdir -p ~/sparkapp/src/main/scala
cd ~/sparkapp

关键的 build.sbt 文件配置：

name := "TopN"
version := "1.0"
scalaVersion := "2.12.15"  // 必须与Spark内置Scala版本匹配
libraryDependencies += "org.apache.spark" %% "spark-core" % "3.3.0"

版本匹配要点 ：

• Spark 3.x需要Scala 2.12+
• 使用 %% 而不是 % 可以自动处理Scala版本后缀
• 可以通过 spark-shell --version 查看Spark内置的Scala版本

4. 编写Spark TopN程序

以下是完整的 TopN.scala 代码，包含详细的错误处理和日志控制：

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.log4j.{Level, Logger}

object TopN {
  def main(args: Array[String]): Unit = {
    // 1. 配置SparkContext
    val conf = new SparkConf()
      .setAppName("PaymentTopN")
      .setMaster("local[*]")  // 本地模式使用所有核心
    
    val sc = new SparkContext(conf)
    
    // 2. 设置日志级别（避免过多调试输出）
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("akka").setLevel(Level.ERROR)
    
    try {
      // 3. 读取HDFS文件（注意替换your-hostname）
      val hdfsPath = "hdfs://your-hostname:9000/example/file*.txt"
      val lines = sc.textFile(hdfsPath, 2)
      
      // 4. 数据处理流水线
      val top5 = lines
        .filter(_.trim.nonEmpty)  // 过滤空行
        .map(_.split(","))        // 分割字段
        .filter(_.length == 4)    // 确保有4个字段
        .map(fields => fields(2).toInt)  // 提取payment字段
        .sortBy(-_)               // 降序排序
        .take(5)                  // 取前5
        
      // 5. 打印结果
      println("\nTop 5 Payment Amounts:")
      top5.zipWithIndex.foreach { case (amount, idx) =>
        println(s"${idx + 1}. $amount")
      }
      
    } catch {
      case e: Exception =>
        println(s"Error occurred: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      // 6. 关闭SparkContext
      sc.stop()
    }
  }
}

代码关键点解析 ：

1. 资源管理 ：使用try-finally确保SparkContext始终被关闭
2. 数据验证 ：多重过滤确保数据质量
3. 性能优化 ：设置合理的分区数（textFile的第二个参数）
4. 错误处理 ：捕获并打印异常信息

5. 构建与提交应用

完成代码编写后，按照以下步骤构建和运行：

# 进入项目目录
cd ~/sparkapp

# 打包（首次运行会下载依赖，较慢）
sbt package

# 提交到Spark（假设Spark安装在/opt/spark）
/opt/spark/bin/spark-submit \
  --class "TopN" \
  --master local[*] \
  target/scala-2.12/topn_2.12-1.0.jar

常见打包问题解决 ：

1. 依赖下载慢 ： 在 ~/.sbt/ 下创建 repositories 文件：

   [repositories]
   local
   maven-central: https://maven.aliyun.com/repository/central
   

2. 版本冲突 ： 在 sbt 中运行 dependencyTree 查看依赖关系：

   sbt dependencyTree
   

3. 内存不足 ： 修改 sbtopts 文件增加内存：

   -Xms1024M
   -Xmx2048M
   

6. 高级技巧与优化

当基本功能实现后，可以考虑以下优化：

1. 使用DataFrame API（推荐）

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("TopN").getOrCreate()

val df = spark.read
  .option("header", "false")
  .csv("hdfs://your-hostname:9000/example/file*.txt")
  .toDF("orderid", "userid", "payment", "productid")

df.createOrReplaceTempView("payments")

val top5 = spark.sql("""
  SELECT CAST(payment AS INT) as amount 
  FROM payments 
  ORDER BY amount DESC 
  LIMIT 5
""")

top5.show()

2. 参数化N值

修改代码接受命令行参数：

object TopN {
  def main(args: Array[String]): Unit = {
    val N = if (args.length > 0) args(0).toInt else 5
    // ...其余代码...
    .take(N)  // 替换原来的take(5)
  }
}

提交时传递参数：

spark-submit --class "TopN" target/...jar 10  # 获取Top 10

3. 结果保存到HDFS

top5
  .map(_.toString)
  .saveAsTextFile("hdfs://your-hostname:9000/output/top5")

7. 真实场景问题排查

在实际企业环境中，你可能会遇到：

问题1 ：数据量太大导致OOM

解决方案 ：

• 增加executor内存： --executor-memory 4G
• 减少并行度： .coalesce(10)
• 使用磁盘缓存： .persist(StorageLevel.DISK_ONLY)

问题2 ：数据倾斜

诊断方法 ：

val paymentDist = df
  .groupBy("payment")
  .count()
  .orderBy($"count".desc)
paymentDist.show(10)

���决方案 ：

• 加盐处理
• 两阶段聚合

问题3 ：HDFS连接超时

配置调整 ：

<!-- 在core-site.xml中增加 -->
<property>
  <name>fs.hdfs.impl.disable.cache</name>
  <value>true</value>
</property>
<property>
  <name>ipc.client.connect.timeout</name>
  <value>30000</value>
</property>