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

从两个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是另一个"坑"密集区。以下是经过验证的安装步骤：

下载正确的版本：
- 访问 sbt官网
- 选择1.9.x系列版本（与Spark 3.x兼容性最好）
安装过程：

# 创建安装目录 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() } } }

代码关键点解析：

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

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

常见打包问题解决：

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

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

版本冲突：在sbt中运行dependencyTree查看依赖关系：
```
sbt dependencyTree
```
内存不足：修改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>

企业官网建设流程全解析