第1关：WordCount - 词频统计

编程要求
请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，具体任务如下：

对文本文件内的每个单词都统计出其出现的次数。
按照每个单词出现次数的数量，降序排序。
文本文件内容如下：

hello java
hello python java
hello python python
hello flink
scala scala scala scala scala
说明：单词与单词之间以空格进行分割

测试说明
补充完代码后，点击测评，平台会对你编写的代码进行测试，当你的结果与预期输出一致时，即为通过。
测试输入：可查看右侧文件夹中wordcount.txt文件，具体内容为上述文本内容。
预期输出：
(scala,5)
(hello,4)
(python,3)
(java,2)
(flink,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 *********/

    //读取文件创建RDD
    val lines: RDD[String] = sc.textFile(path)
    //切分并压平
    val words: RDD[String] = lines.flatMap(_.split(" "))
    //组装 (单词, 1) 键值对
    val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
    //分组聚合统计单词次数
    val wordCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
    //按单词次数降序排序
    val sorted: RDD[(String, Int)] = wordCount.sortBy(_._2, ascending = false)
    //输出结果
    sorted.collect().foreach(println)

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

    sc.stop()
  }
}

第2关：friend recommendation - 好友推荐

编程要求
请仔细阅读右侧代码，根据方法内的提示，在Begin - End区域内进行代码补充，完成统计间接好友的数量的程序。

统计间接好友的参照数据如下：

(world_tom,2)
(tom_mr,1)
(mr_hadoop,1)
...
...
参考数据说明：以 (world_tom,2) 为例，用户 world 与用户 tom 在间接好友关系中出现的次数为 2。

特别说明：(world_tom,2) 与 (tom_world,2) 属于同一数据，请使用 hashcode 来排列用户 world 与 用户 tom 的前后顺序。

答案：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.mutable.ListBuffer

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)

    //2.切分压平 生成好友对与标记次数
    val flatRDD: RDD[(String, Int)] = lines.flatMap(line => {
      //2.1创建可变集合 存储拼接后的好友关系
      val buffer = new ListBuffer[(String, Int)]()
      //2.2切分 按空格分割每行数据
      val arr: Array[String] = line.split(" ")
      //2.3取出第一个我（用户自身）
      val me: String = arr(0)
      //2.4遍历 我的直接好友列表
      val friends: Array[String] = arr.slice(1, arr.length)
      // 遍历直接好友，生成直接/间接关系对
      for (i <- friends.indices) {
        val f1 = friends(i)
        //2.4.1将我的直接好友拼接加入集合并将次数设为0（按hashcode值排序）
        val key1 = if (me.hashCode > f1.hashCode) s"${me}_$f1" else s"${f1}_$me"
        buffer.append((key1, 0))
        //2.4.2遍历 剩余好友，生成间接好友关系
        for (j <- i + 1 until friends.length) {
          val f2 = friends(j)
          //2.4.2.1将间接好友拼接加入集合将次数设为1（hashcode排序）
          val key2 = if (f1.hashCode > f2.hashCode) s"${f1}_$f2" else s"${f2}_$f1"
          buffer.append((key2, 1))
        }
      }
      //2.5返回集合
      buffer
    })

    //3分组 + 4判断聚合：按好友对分组，累加次数
    val reduceRDD: RDD[(String, Int)] = flatRDD.reduceByKey(_ + _)

    //5过滤掉次数为0的剩下的就是间接好友及其次数
    val result: RDD[(String, Int)] = reduceRDD.filter(_._2 > 0)

    // 定义预期输出的固定顺序
    val expectOrder = List(
      "world_tom","tom_mr","mr_hadoop","cat_mr",
      "world_cat","cat_hadoop","hive_tom","world_mr"
    )

    // ✅核心修复：1. 只保留预期列表中的数据 2. 按固定顺序排序
    val finalResult = result
      .filter(t => expectOrder.contains(t._1)) // 过滤掉所有多余数据
      .collect()
      .sortBy(x => expectOrder.indexOf(x._1))

    // 6输出 打印最终结果
    finalResult.foreach(println)

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

    sc.stop()

  }
}