输出两道 Spark Streaming相关的程序题并给出答案

题目1：

使用Spark Streaming实现一个简单的实时词频统计程序。输入数据为一个DStream，其中每个RDD包含多个单词，输出数据为一个DStream，其中每个RDD包含每个单词出现的次数。

答案：

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.SparkConf

object WordCount {
  def main(args: Array[String]) {
    // 创建SparkConf对象
    val conf = new SparkConf().setAppName("WordCount").setMaster("local[*]")
    // 创建StreamingContext对象，每秒钟处理一次数据
    val ssc = new StreamingContext(conf, Seconds(1))
    // 从socket读取数据
    val lines = ssc.socketTextStream("localhost", 9999)
    // 切分每行数据为单词
    val words = lines.flatMap(_.split(" "))
    // 统计每个单词的出现次数
    val wordCounts = words.map((_, 1)).reduceByKey(_ + _)
    // 打印结果
    wordCounts.print()
    // 启动StreamingContext
    ssc.start()
    // 等待StreamingContext停止
    ssc.awaitTermination()
  }
}

题目2：

使用Spark Streaming实现一个实时的广告点击率统计程序。输入数据为一个DStream，其中每个RDD包含多个广告点击事件，每个事件包含广告ID和点击次数，输出数据为一个DStream，其中每个RDD包含每个广告的点击率。

答案：

import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.SparkConf

case class AdClick(adId: String, clickCount: Int)

object AdClickRate {
  def main(args: Array[String]) {
    // 创建SparkConf对象
    val conf = new SparkConf().setAppName("AdClickRate").setMaster("local[*]")
    // 创建StreamingContext对象，每秒钟处理一次数据
    val ssc = new StreamingContext(conf, Seconds(1))
    // 从socket读取数据
    val lines = ssc.socketTextStream("localhost", 9999)
    // 将每行数据转换为AdClick对象
    val adClicks = lines.map(line => {
      val tokens = line.split(",")
      AdClick(tokens(0), tokens(1).toInt)
    })
    // 统计每个广告的总点击次数和总展示次数
    val adCounts = adClicks.map(ad => (ad.adId, (ad.clickCount, 1)))
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
    // 计算每个广告的点击率
    val adClickRates = adCounts.mapValues(counts => counts._1.toDouble / counts._2.toDouble)
    // 打印结果
    adClickRates.print()
    // 启动StreamingContext
    ssc.start()
    // 等待StreamingContext停止
    ssc.awaitTermination()
  }
}
``