用Spark解决三道经典数据处理题：去重/求平均/HDFS统计（附完整Scala代码）

用Spark解决三道经典数据处理题：去重/求平均/HDFS统计（附完整Scala代码）

在大数据领域，Spark已经成为处理海量数据的首选框架之一。无论是面试还是实际业务场景，掌握Spark的核心操作都是数据工程师的必备技能。本文将带你通过三个典型场景，深入理解Spark的基础开发技巧，每个案例都包含完整的Scala代码实现和性能优化建议。

1. 数据去重实战：合并两个数据集并去除重复项

数据去重是ETL过程中的常见需求。假设我们有两个用户行为日志文件A和B，需要合并后去除完全相同的记录。以下是完整的Scala实现：

import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.rdd.RDD object DataDeduplication { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("DataDeduplication") val sc = new SparkContext(conf) // 读取两个输入文件 val fileA = sc.textFile("hdfs://path/to/fileA") val fileB = sc.textFile("hdfs://path/to/fileB") // 合并并去重 val combined = fileA.union(fileB).distinct() // 保存结果 combined.saveAsTextFile("hdfs://path/to/output") sc.stop() } }

性能优化建议：

• 对于超大规模数据集，可以增加分区数：.repartition(100)
• 如果数据有特定键值，使用.reduceByKey比.distinct更高效
• 考虑使用.persist()缓存中间结果，避免重复计算

注意：实际生产环境中，distinct操作可能会引起数据倾斜，需要根据数据特点进行优化。

2. 学生成绩统计：多科目平均分计算

教育数据分析中经常需要计算学生的平均成绩。下面是一个处理多科目成绩的Spark实现：

import org.apache.spark.{SparkConf, SparkContext} object AverageScoreCalculator { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("AverageScoreCalculator") val sc = new SparkContext(conf) // 读取所有科目成绩文件 val scores = sc.textFile("hdfs://path/to/scores/*") // 转换为(学生姓名, (分数, 1))的元组 val scorePairs = scores.map { line => val parts = line.split(" ") (parts(0), (parts(1).toDouble, 1)) } // 计算总分和科目数 val scoreSums = scorePairs.reduceByKey { case ((sum1, count1), (sum2, count2)) => (sum1 + sum2, count1 + count2) } // 计算平均分并格式化 val averages = scoreSums.mapValues { case (sum, count) => f"${sum / count}%1.2f".toDouble } // 按平均分降序排序 val sortedAverages = averages.sortBy(-_._2) sortedAverages.saveAsTextFile("hdfs://path/to/averages") sc.stop() } }

关键点解析：

1. 使用reduceByKey高效聚合数据
2. 保留计数信息以准确计算平均值
3. 格式化输出保留两位小数
4. 最终结果按分数排序，便于分析

3. HDFS文件统计：行数与大小分析

监控HDFS文件状态是运维常见任务。以下代码展示了如何统计HDFS文件的行数和大小：

import org.apache.spark.{SparkConf, SparkContext} import org.apache.hadoop.fs.{FileSystem, Path} object HDFSFileAnalyzer { def main(args: Array[String]): Unit = { val conf = new SparkConf().setAppName("HDFSFileAnalyzer") val sc = new SparkContext(conf) val filePath = "hdfs://path/to/input/file" // 统计行数 val lines = sc.textFile(filePath) val lineCount = lines.count() // 获取文件大小 val fs = FileSystem.get(sc.hadoopConfiguration) val fileSize = fs.getContentSummary(new Path(filePath)).getLength println(s"文件行数: $lineCount") println(s"文件大小: ${fileSize / (1024 * 1024)} MB") sc.stop() } }

扩展功能：

• 可以添加文件修改时间检查
• 支持递归统计目录下所有文件
• 添加文件格式验证逻辑

4. 生产环境最佳实践

在实际项目中应用这些技术时，还需要考虑以下因素：

配置优化参数：

常见问题排查：

• 内存不足：增加executor内存或减少分区数
• 数据倾斜：
  • 使用sample检查数据分布
  • 对倾斜键单独处理
• 性能瓶颈：
  • 检查DAG执行计划
  • 合理使用缓存策略

代码质量建议：

1. 添加完善的日志记录
2. 实现参数化配置
3. 添加单元测试
4. 考虑使用Spark SQL替代RDD操作
5. 使用DataFrame API获得更好的优化效果