Spark实战：3个真实场景下的数据处理案例详解（去重、统计、求平均）

Spark实战3个真实场景下的数据处理案例详解去重、统计、求平均当你面对海量数据时传统单机处理方式往往力不从心。Spark作为分布式计算框架能够高效处理TB甚至PB级数据。本文将带你深入三个实际工作中最常见的数据处理场景通过完整代码示例和原理剖析掌握Spark核心操作技巧。1. 数据去重实战合并多源数据并剔除重复项电商平台每天会产生大量用户行为日志这些日志通常分散在不同系统中。我们需要合并这些数据并去除重复记录以便后续分析用户行为路径。1.1 数据准备与问题分析假设我们有两份用户浏览记录文件A记录上午的用户访问文件B记录下午的用户访问两份数据可能存在部分重叠即同一用户在上午和下午访问了相同商品。我们需要合并这两份数据同时确保每条记录唯一。// 示例输入数据 val dataA Seq( 20230501_10:00 user1 productA, 20230501_10:15 user2 productB, 20230501_11:30 user1 productC ) val dataB Seq( 20230501_14:00 user1 productA, // 与dataA重复 20230501_15:45 user3 productD, 20230501_16:20 user2 productE )1.2 完整去重解决方案使用union合并RDD后通过distinct操作实现去重import org.apache.spark.{SparkConf, SparkContext} 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 uniqueData fileA.union(fileB).distinct() // 保存结果 uniqueData.saveAsTextFile(hdfs://path/to/output) sc.stop() } }注意对于超大数据集distinct()可能导致shuffle操作可通过调整分区数优化性能1.3 进阶优化技巧当处理超大规模数据时可以考虑以下优化方案方案对比表方法优点缺点适用场景distinct()简单直接全量shuffle中小规模数据reduceByKey可控制shuffle量需要构造键值对可分组数据布隆过滤器内存效率高存在误判率近似去重// 使用reduceByKey实现去重需确保每行数据作为key val keyValuePairs fileA.union(fileB).map(line (line, null)) val uniqueData keyValuePairs.reduceByKey((a,b) a).keys2. 行数统计从基础到高级应用行数统计看似简单但在实际业务中我们往往需要更复杂的统计逻辑比如按条件过滤后的计数或者分布式环境下的精确统计。2.1 基础行数统计实现最基本的行数统计使用count()方法val logFile sc.textFile(hdfs://path/to/largefile.log) val totalLines logFile.count() println(s文件总行数: $totalLines)2.2 带条件过滤的统计实际业务中我们经常需要统计满足特定条件的行数// 统计包含ERROR关键字的日志行数 val errorLogs logFile.filter(line line.contains(ERROR)).count() // 多条件统计 val stats Map( total - logFile.count(), errors - logFile.filter(_.contains(ERROR)).count(), warnings - logFile.filter(_.contains(WARNING)).count() ) stats.foreach { case (k, v) println(s$k: $v) }2.3 大规模数据统计优化当数据量极大时count()操作可能较慢。如果只需要近似结果可以使用采样统计// 采样0.1%的数据估算总行数 val sampleRatio 0.001 val approxCount logFile.sample(withReplacement false, sampleRatio).count() / sampleRatio性能对比测试结果数据集大小count()耗时采样统计耗时误差率100GB45s3s±0.2%1TB6min8s±0.5%3. 求平均值多维分析与性能考量计算平均值是数据分析中最常见的操作之一。我们将通过学生成绩分析的案例展示如何处理多源数据并计算综合平均值。3.1 基础平均值计算假设有三个学科的成绩文件我们需要计算每个学生的平均分val mathScores sc.textFile(hdfs://path/to/math.txt) .map(line { val parts line.split( ) (parts(0), parts(1).toDouble) // (学生姓名, 分数) }) val physicsScores sc.textFile(hdfs://path/to/physics.txt) .map(line { val parts line.split( ) (parts(0), parts(1).toDouble) }) val chemistryScores sc.textFile(hdfs://path/to/chemistry.txt) .map(line { val parts line.split( ) (parts(0), parts(1).toDouble) })3.2 多数据集合并与计算使用union合并所有成绩然后通过reduceByKey计算总分和科目数// 合并所有成绩 val allScores mathScores.union(physicsScores).union(chemistryScores) // 计算每个学生的总分和科目数 val scoreSumAndCount allScores .mapValues(score (score, 1)) // 转换为(分数, 1)的元组 .reduceByKey((a, b) (a._1 b._1, a._2 b._2)) // 计算平均分并格式化输出 val averageScores scoreSumAndCount.map { case (name, (sum, count)) val avg sum / count f$name: $avg%.2f // 保留两位小数 } averageScores.saveAsTextFile(hdfs://path/to/output/averages)3.3 处理数据倾斜问题当某些key的数据量远大于其他key时比如某些学生参加了更多考试会导致数据倾斜。解决方案// 方案1增加分区数 val allScores mathScores.union(physicsScores).union(chemistryScores) .repartition(200) // 根据数据量调整分区数 // 方案2使用salting技术 val saltedScores allScores.map { case (name, score) val salt scala.util.Random.nextInt(10) // 0-9的随机数 (s$name-$salt, score) } // 计算后再合并结果 val saltedAverages saltedScores .mapValues(score (score, 1)) .reduceByKey((a, b) (a._1 b._1, a._2 b._2)) .map { case (saltedName, (sum, count)) val name saltedName.split(-)(0) (name, (sum, count)) } .reduceByKey((a, b) (a._1 b._1, a._2 b._2)) .map { case (name, (sum, count)) f$name: ${sum / count}%.2f }4. 生产环境最佳实践将上述案例应用到生产环境时还需要考虑以下关键因素4.1 性能调优参数重要配置参数表参数推荐值说明spark.executor.memory8G-16G根据数据量调整spark.executor.cores4-8每个executor的CPU核心数spark.dynamicAllocation.enabledtrue启用动态资源分配spark.sql.shuffle.partitions200-400shuffle操作的分区数spark.default.parallelism同shuffle分区默认并行度4.2 监控与调试添加监控代码跟踪作业执行// 在SparkConf中添加监控配置 val conf new SparkConf() .setAppName(DataProcessingApp) .set(spark.metrics.conf, /path/to/metrics.properties) .set(spark.eventLog.enabled, true) .set(spark.eventLog.dir, hdfs://path/to/event-logs) // 在代码中添加性能标记 val startTime System.nanoTime() // ...数据处理逻辑... val duration (System.nanoTime() - startTime) / 1e9d println(f作业执行时间: $duration%.2f 秒) // 打印RDD血统信息 println(toDebugString)4.3 错误处理与数据校验健壮的生产代码应该包含完善的错误处理// 安全读取和解析数据 val safeScores sc.textFile(hdfs://path/to/scores.txt).flatMap { line try { val parts line.split( ) if (parts.length 2) { Some((parts(0), parts(1).toDouble)) } else { None } } catch { case e: NumberFormatException println(s解析失败: $line) None case e: Exception println(s未知错误: $line) None } } // 验证数据质量 val validRecords safeScores.count() val invalidRecords safeScores.filter(_ null).count() println(s有效记录: $validRecords, 无效记录: $invalidRecords)在完成这三个案例的实践后建议尝试将它们组合起来解决更复杂的问题。比如先对原始数据去重然后统计各类事件的数量最后计算某些指标的均值。这种端到端的处理流程更能体现Spark在实际工作中的价值。