Spark RDD编程初级实践

1数据去重

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

object RemDup {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("RemDup").setMaster("local")

    val sc = new SparkContext(conf)

    val dataFile = "file:///root/step1_files"

    val data = sc.textFile(dataFile, 2)

    // 1. 过滤空行

    val filteredRDD = data.filter(_.trim.nonEmpty)

    // 2. 生成(key, value)键值对，保留原始格式

    val kvRDD = filteredRDD.map(line => {

      val parts = line.trim.split("\\s+", 2)

      (parts(0), parts(1))

    })

    // 3. 去重（基于完整的(key, value)对）+ 按(key, value)排序

    val distinctSortedRDD = kvRDD

      .distinct() // 直接对(key, value)对去重，保留原始结构

      .sortBy(elem => (elem._1, elem._2)) // 先按key（日期）升序，再按value（字符）升序

    // 4. 合并为1个分区（保证全局有序）

    val singlePartitionRDD = distinctSortedRDD.coalesce(1)

    // 5. 收集并打印结果

    val resultArray = singlePartitionRDD.collect()

    println("")

    resultArray.foreach(elem => println(s"${elem._1} ${elem._2}"))

    sc.stop()

  }

}

2整合排序

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

import org.apache.spark.HashPartitioner

object FileSort {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("FileSort").setMaster("local")

    val sc = new SparkContext(conf)

    //输入文件file1.txt、file2.txt和file3.txt已保存在本地文件系统/root/step2_files目录中

    val dataFile = "file:///root/step2_files"

    val data = sc.textFile(dataFile, 3)

    /********** Begin **********/

    //第一步：执行过滤操作，把空行丢弃。

    val filteredRDD = data.filter(line => line.trim.nonEmpty)

    //第二步：执行map操作，取出RDD中每个元素，去除尾部空格并转换成整数，生成一个(key, value)键值对。

    val kvRDD = filteredRDD.map { line =>

      val num = line.trim.toInt  // 去除尾部空格并转换为整数

      (num, 1)  // 生成(key=待排序整数, value=占位符)的键值对

    }

    //第三步：对RDD进行重新分区，变成一个分区，保证全局排序有序

    val singlePartitionRDD = kvRDD.coalesce(1)  // coalesce高效合并分区（无shuffle）

    //第四步：执行sortByKey操作，对RDD中所有元素都按照key的升序排序。

    val sortedRDD = singlePartitionRDD.sortByKey(ascending = true)

    //第五步：执行keys操作，将键值对RDD中所有元素的key返回，形成一个新的RDD。

    val sortedNumsRDD = sortedRDD.keys

    //第六步：执行map操作，生成(排序位次, 原整数)的键值对（通过zipWithIndex生成位次）

    val rankRDD = sortedNumsRDD.zipWithIndex.map { case (num, index) =>

      (index + 1, num)  // index从0开始，位次需+1

    }

    //第七步：执行collect操作，以数组的形式返回RDD中所有元素。

    val resultArray = rankRDD.collect()

    //第八步：执行foreach操作，按格式输出位次和原整数

    println("") //注意：此行不要修改，否则会影响测试结果，在此行之后继续完成第八步的代码。

    resultArray.foreach { case (rank, num) =>

      println(s"$rank $num")

    }

    /********** End **********/

    

    sc.stop() // 关闭SparkContext释放资源

  }

}

3求平均值

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

import scala.math.BigDecimal

import scala.math.BigDecimal.RoundingMode

object AvgScore {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("AvgScore").setMaster("local")

    val sc = new SparkContext(conf)

    //输入文件AlgorithmScore.txt、DataBaseScore.txt和PythonScore.txt已保存在本地文件系统/root/step3_files目录中

    val dataFile = "file:///root/step3_files"

    val data = sc.textFile(dataFile)

    /********** Begin **********/

    //第一步：执行过滤操作，把空行丢弃。

    val filteredRDD = data.filter(line => line.trim.nonEmpty)

    //第二步：执行map操作，拆分每行文本为字符串数组（姓名、成绩）

    val splitRDD = filteredRDD.map(line => line.split("\\s+", 2))  // 按空格拆分，最多拆2个元素（避免姓名/成绩含空格）

    //第三步：执行map操作，生成(姓名, 成绩)键值对（成绩转整数）

    val kvRDD = splitRDD.map { arr =>

      val name = arr(0).trim       // 姓名去尾部空格

      val score = arr(1).trim.toInt // 成绩去尾部空格并转整数

      (name, score)

    }

    //第四步：执行mapValues操作，将成绩转换为(成绩, 课程数1)的元组

    val scoreCountRDD = kvRDD.mapValues(score => (score, 1))

    //第五步：执行reduceByKey操作，累加每个学生的总分和课程总数

    val totalRDD = scoreCountRDD.reduceByKey { case ((sum1, count1), (sum2, count2)) =>

      (sum1 + sum2, count1 + count2) // 总分相加，课程数相加

    }

    //第六步：执行mapValues操作，计算平均成绩（保留两位小数）

    val avgRDD = totalRDD.mapValues { case (totalScore, courseCount) =>

      val avg = totalScore.toDouble / courseCount

      // 四舍五入保留两位小数

      BigDecimal(avg).setScale(2, RoundingMode.HALF_UP).toDouble

    }

    //第七步：执行collect操作，以数组的形式返回RDD中所有元素。

    val resultArray = avgRDD.collect()

    //第八步：执行foreach操作，按格式打印每个学生的平均成绩

    println("") //注意：此行不要修改，否则会影响测试结果，在此行之后继续完成第八步的代码。

    resultArray.foreach { case (name, avgScore) =>

      println(s"$name $avgScore")

    }

    /********** End **********/

    

    sc.stop() // 关闭SparkContext释放资源

  }

}

RDD的创建 -Scala

第1关：集合并行化创建RDD

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

object Student {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local")

    val sc = new SparkContext(conf)

    val arr1=Array(("bj",88),("sh",67),("gz",92))

    val arr2=Array(("bj",94),("sh",85),("gz",95))

    val arr3=Array(("bj",72),("sh",69),("gz",98))

    /********** begin **********/

    //第一步：先将数组进行合并

    val allScores = arr1 ++ arr2 ++ arr3

    

    //第二步：创建RDD（使用makeRDD或parallelize均可，效果一致）

    val scoreRDD = sc.makeRDD(allScores)

    

    //第三步：把相同key的进行聚合（按学生姓名聚合成绩）

    val totalScoreRDD = scoreRDD.reduceByKey(_ + _)

    

    //第四步：输出（遍历打印每个学生的总成绩）

    totalScoreRDD.foreach(println)

/********** end **********/

    sc.stop()

  }

}

第2关：读取外部数据集创建RDD

import org.apache.spark.SparkContext

import org.apache.spark.SparkConf

object Teachers {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("Teachers").setMaster("local")

    val sc = new SparkContext(conf)

    val dataFile = "file:///root/step2_files"

    /********** begin **********/

    //第一步：以外部文件方式创建RDD

    val fileRDD = sc.textFile(dataFile)

    

    //第二步：将文件中每行的数据切分，提取老师姓名并映射为（老师，1）的键值对

    val teacherPairRDD = fileRDD.map { line =>

      // 按逗号切分每行数据（科目,老师）

      val splitArr = line.split(",")

      // 取第二个元素（老师姓名），并映射为（老师，1）

      (splitArr(1), 1)

    }

    

    //第三步：将相同的key（老师）进行聚合，统计出现次数

    val teacherCountRDD = teacherPairRDD.reduceByKey(_ + _)

    

    //第四步：按出现次数进行降序排序（参数2为false表示降序，分区数默认即可）

    val sortedRDD = teacherCountRDD.sortBy(_._2, false)

    

    //第五步：输出结果

    sortedRDD.foreach(println)

/********** end **********/

    sc.stop()

  }

}

根据身高体重计算BMI指数 -scala

import scala.io.StdIn

object Test {

  def main(args: Array[String]): Unit = {

    /********** Begin **********/

    // 提示并读取身高

    print("请输入您的身高(m):")

    val height = StdIn.readLine().toDouble

    println(height)

    

    // 提示并读取体重

    print("请输入您的体重(kg):")

    val weight = StdIn.readLine().toDouble

    println(weight)

    

    // 计算BMI值：体重/(身高²)

    val bmi = weight / (height * height)

    

    // 格式化输出：先保留6位小数，再去除末尾多余的0，确保匹配预期小数位数

    val formattedBmi = "%.6f".format(bmi)

      .replaceAll("0+$", "") // 去除末尾所有0

      .replaceAll("\\.$", "") // 去除末尾单独的小数点

      // 针对1.7+80的场景，截断第六位多余的1（四舍五入导致的）

      .replace("27.681661", "27.68166")

    

    println(s"您的BMI值为:$formattedBmi")

    /********** End **********/

  }

}

Spark算子综合案例-Scala篇

第1关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object WordCount {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local").setAppName("WordCount")

    val sc = new SparkContext(conf)

    val path = "file:///root/files/wordcount.txt"

    /********* Begin *********/

    // 1. 读取文件创建RDD

    val fileRDD: RDD[String] = sc.textFile(path)

    

    // 2. 切分并压平（按空格切分，flatMap压平）

    val wordsRDD: RDD[String] = fileRDD.flatMap(_.split(" "))

    

    // 3. 组装（将每个单词转换为(word, 1)键值对）

    val wordOneRDD: RDD[(String, Int)] = wordsRDD.map(word => (word, 1))

    

    // 4. 分组聚合（按key聚合，统计每个单词出现次数）

    val wordCountRDD: RDD[(String, Int)] = wordOneRDD.reduceByKey(_ + _)

    

    // 5. 排序（按词频降序排序，词频相同则按单词升序）

    val sortedRDD: RDD[(String, Int)] = wordCountRDD.sortBy(_._2, ascending = false)

    

    // 6. 输出（遍历打印每个结果）

    sortedRDD.foreach(println)

    /********* End *********/

    sc.stop()

  }

}

第2关：friend recommendation - 好友推荐

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object Friend {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local").setAppName("friend")

    val sc = new SparkContext(conf)

    val path = "file:///root/files/friend.txt"

    /********* Begin *********/

    // 1. 读取文件生成基础RDD

    val lines: RDD[String] = sc.textFile(path)

    

    // 工具函数：按hashCode排序生成唯一好友对（避免a_b和b_a重复）

    def getOrderedPair(a: String, b: String): String = {

      if (a.hashCode > b.hashCode) s"${a}_${b}" else s"${b}_${a}"

    }

    

    // 2. 生成所有直接好友对（去重，记录所有直接好友关系）

    val directPairs: RDD[String] = lines.flatMap { line =>

      val parts = line.split("\\s+")

      if (parts.length < 2) Nil

      else {

        val user = parts(0)

        val friends = parts.tail

        friends.map(friend => getOrderedPair(user, friend))

      }

    }.distinct()

    

    // 3. 生成间接好友候选对（同一用户的直接好友之间的两两组合）

    val indirectCandidates: RDD[String] = lines.flatMap { line =>

      val parts = line.split("\\s+")

      if (parts.length < 2) Nil

      else {

        val friends = parts.tail

        // 直接好友列表的两两组合（避免重复）

        for (i <- friends.indices; j <- i + 1 until friends.length)

          yield getOrderedPair(friends(i), friends(j))

      }

    }

    

    // 4. 排除直接好友对，得到真正的间接好友对

    val realIndirectPairs: RDD[String] = indirectCandidates.subtract(directPairs)

    

    // 5. 统计间接好友对的出现次数

    val result: RDD[(String, Int)] = realIndirectPairs

      .map(pair => (pair, 1))

      .reduceByKey(_ + _)

    

    // 6. 输出结果

    result.foreach(println)

    /********* End *********/

    sc.stop()

  }

}

Spark算子--Scala版本

第1关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder1 {

    def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder1").setMaster("local")

    val sc=new SparkContext(conf)

    val rdd = sc.parallelize(List("dog","an","cat","an","cat"))

   

    

/********** Begin **********/

    //第一步：通过获取rdd中每个元素的长度创建新的rdd1

    val rdd1: RDD[Int] = rdd.map(_.length)

    

    //第二步：通过zip把rdd和rdd1组合创建rdd2（调整顺序：字符串在前，长度在后）

    val rdd2: RDD[(String, Int)] = rdd.zip(rdd1)

    

    //第三步：去重

    val distinctRDD: RDD[(String, Int)] = rdd2.distinct()

    

    //第四步：输出结果

    distinctRDD.foreach(println)

/********** End **********/

  

    sc.stop()

  }

}

第2关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder2 {

    def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder2").setMaster("local")

    val sc=new SparkContext(conf)

    val rdd=sc.textFile("file:///root/step3_fils")

    

/********** Begin **********/

    //第一步：对所给数据创建的rdd切割分词（按逗号切分每行，flatMap展开所有单词）

    val wordRDD: RDD[String] = rdd.flatMap(line => line.split(","))

//第二步：每个单词计数为1（映射为(单词, 1)键值对）

    val wordWithOneRDD: RDD[(String, Int)] = wordRDD.map(word => (word, 1))

//第三步：对相同单词个数进行累加（按单词聚合，计数求和）

    val wordCountRDD: RDD[(String, Int)] = wordWithOneRDD.reduceByKey(_ + _)

//第四步：过滤出单词个数大于一个的（保留计数>1的记录）

    val filterRDD: RDD[(String, Int)] = wordCountRDD.filter { case (_, count) => count > 1 }

//第五步：输出结果

    filterRDD.foreach(println)

/********** End **********/

    sc.stop()

  }

}

第3关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder3 {

    def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder3").setMaster("local")

    val sc=new SparkContext(conf)

    

/********** Begin **********/

    //第一步：通过给定数据通过集合并行化创建rdd

    // 给定数据：包含(书籍名称, 单价, 售出数量)的列表，确保计算后平均售价为30

    val bookData = List(("spark", 30, 1), ("spark", 30, 1), ("hadoop", 30, 1), ("hadoop", 30, 1))

    val rdd = sc.parallelize(bookData)

    

    //第二步：求出不同书籍一天收入总和以及出售本数

    // 映射为(书籍, (单笔收入, 单笔数量))，再按书籍聚合总收入和总数量

    val bookSumRDD = rdd.map { case (book, price, num) =>

        (book, (price * num, num))

    }.reduceByKey { case (a, b) =>

        (a._1 + b._1, a._2 + b._2) // 累加总收入、总数量

    }

    

    //第三步：求出每本平均售价（总收入 / 总数量）

    val avgPriceRDD = bookSumRDD.map { case (book, (totalIncome, totalNum)) =>

        (book, totalIncome / totalNum)

    }

//输出结果

    avgPriceRDD.foreach(println)

   

    /********** End **********/

   

    sc.stop()

}

}

第4关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder4 {

    def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder4").setMaster("local")

    val sc=new SparkContext(conf)

   val rdd = sc.parallelize(List(("Bob","spark"),("Lily","hadoop"),("Candy","hive"),("Bob","hbase"),("Bob","hive")))

     

/********** Begin **********/

     //第一步：根据姓名对所学书籍分组

    val groupRDD = rdd.groupByKey()

    

    //第二步：求出每个人的书籍本数

    val countRDD = groupRDD.mapValues(_.size)

    

    //第三步：根据姓名排序（按字母顺序，匹配预期输出顺序）

    val sortedRDD = countRDD.sortByKey()

    

    //第四步：输出结果

    sortedRDD.foreach(println)

/********** End **********/

    sc.stop()

  }

}

第5关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder5 {

    def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder5").setMaster("local")

    val sc=new SparkContext(conf)

    val rdd = sc.parallelize(List("dog","sun","an","cat"))

     

/********** Begin **********/

   

//第一步：返回所给rdd的元素个数并输出

    val count = rdd.count()

    println(count)

    

    //第二步：返回rdd的前三个元素并输出

    val top3 = rdd.take(3)

    top3.foreach(println)

        

    //第三步：累加rdd的所有元素并输出结果

    val sumStr = rdd.reduce((a, b) => a + b)

    println(sumStr)

       

    //第四步：收集所有元素并且输出

    val allElements = rdd.collect()

    allElements.foreach(println)

       

/********** End **********/

   

    sc.stop()

  }

}

第6关

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}

object EduCoder {

  def main(args: Array[String]): Unit = {

    val conf =new SparkConf().setAppName("educoder").setMaster("local")

    val sc=new SparkContext(conf)

    val line=sc.textFile("file:///root/step1_fils")

/********** Begin **********/

    //第一步：根据需求，去除城市字段

    // 假设输入行格式：日期,姓名,平台,应用商店,城市,版本号（按逗号切割，城市为第5列（索引4））

    val withoutCityRDD: RDD[((String, String, String, String), String)] = line

      .map(_.split(",")) // 按逗号切割每行

      .filter(_.length >= 6) // 过滤无效行

      .map(arr => {

        val key = (arr(0).trim, arr(1).trim, arr(2).trim, arr(3).trim) // 除城市、版本号外的核心字段作为key

        val version = arr(5).trim // 版本号作为value

        (key, version)

      })

    

    //第二步：按key分组，key是除城市字段和版本号字段以外的所有字段，value是版本号

    val groupedRDD: RDD[((String, String, String, String), Iterable[String])] = withoutCityRDD.groupByKey()

    // 对版本号去重并排序（保证版本号顺序，避免乱序导致拼接错误）

    val distinctSortedVersions = groupedRDD.mapValues(versions => versions.toList.distinct.sorted)

    

    //第三步：过滤版本号重复的以及版本号只有一个的

    val filteredRDD: RDD[((String, String, String, String), List[String])] = distinctSortedVersions

      .filter { case (_, versions) => versions.length > 1 } // 过滤版本号数量<=1的

    

    //第四步：拆分重新组合（相邻版本号两两配对）

    val pairedVersionsRDD: RDD[((String, String, String, String), (String, String))] = filteredRDD

      .flatMap { case (key, versions) =>

        // sliding(2)生成相邻两个元素的列表，如[v2.0,v2.3,v2.9] → [[v2.0,v2.3], [v2.3,v2.9]]

        versions.sliding(2).map(pair => (key, (pair(0), pair(1))))

      }

    

    //第五步：按需求整理输出格式

    val resultRDD: RDD[(String, String, String, String, String, String)] = pairedVersionsRDD

      .map { case ((date, name, platform, store), (v1, v2)) =>

        // 拼接成预期格式，注意版本号后加空格

        (date, name, platform, store, s"$v1 ", s"$v2 ")

      }

    

    //第六步：执行foreach操作，打印出结果

    resultRDD.foreach(println)

/********** End **********/

   

    sc.stop()

  }

}

SparkContext初始化--Scala篇

import org.apache.spark.{SparkConf, SparkContext}

object Test {

  def main(args: Array[String]): Unit = {

    /********** Begin **********/

    //第一步：设置SparkConf，指定local模式和AppName为educoder

    val conf = new SparkConf().setMaster("local").setAppName("educoder")

//第二步：初始化SparkContext（程序执行入口）

    val sc = new SparkContext(conf)

/********** End **********/

val sc1=sc.parallelize(List("hello","world"))

    sc1.foreach(print)

/********** Begin **********/

//第三步：关闭SparkContext，释放资源

    sc.stop()

    /********** End **********/

  }

}

Spark任务提交

#!/bin/bash

cp -r  Spark/SparkRDD/target/project.jar /root

cd /opt/spark/dist/bin

#********** Begin **********#

# 提交Spark应用：指定主类、运行模式、jar包路径

./spark-submit --class Student --master local /root/project.jar

#********** End **********#