用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() } }关键点解析使用reduceByKey高效聚合数据保留计数信息以准确计算平均值格式化输出保留两位小数最终结果按分数排序便于分析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. 生产环境最佳实践在实际项目中应用这些技术时还需要考虑以下因素配置优化参数参数名推荐值说明spark.executor.memory8g-16g执行器内存大小spark.driver.memory4g-8g驱动器内存大小spark.default.parallelism集群核数x2-3默认并行度spark.sql.shuffle.partitions200-400shuffle分区数常见问题排查内存不足增加executor内存或减少分区数数据倾斜使用sample检查数据分布对倾斜键单独处理性能瓶颈检查DAG执行计划合理使用缓存策略代码质量建议添加完善的日志记录实现参数化配置添加单元测试考虑使用Spark SQL替代RDD操作使用DataFrame API获得更好的优化效果