Spark Streaming之窗口函数和状态转换函数

流处理主要有3种应用场景：无状态操作、window操作、状态操作。

reduceByKeyAndWindow

import kafka.serializer.StringDecoder import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.SQLContext import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.streaming._ import org.apache.spark.{SparkContext, SparkConf} object ClickStream { def main (args: Array[String]){ // 屏蔽不必要的日志显示在终端上
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) //创建SparkConf对象，设置应用程序的名称，在程序运行的监控界面可以看到名称
    val conf = new SparkConf().setAppName("ClickStream").setMaster("local[*]") val sc = new SparkContext(conf) //此处设置Batch Interval是在Spark Streaming中生成基本Job的时间单位，窗口和滑动时间间隔一定是该Batch Interval的整数倍
    val ssc = new StreamingContext(sc, Seconds(args(0).toLong)) //由于用到了窗口函数，需要复用前面的RDD，必须checkpoint，注意复用的RDD之间是没有任何关系的
    ssc.checkpoint(args(1)) val topics = Set("clickstream")    //所要获取数据在kafka上的主题
    val brokers = "yz4203.hadoop.data.sina.com.cn:19092,yz4202.hadoop.data.sina.com.cn:19092,10.39.4.212:19092,10.39.4.201:19092" val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers) //val offset = "largest" //values: smallest, largest ，控制读取最新的数据，还是旧的数据, 默认值为largest //从Spark1.3开始，我们能够使用如下方式高效地从kafka上获取数据
    val kvsTemp = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) val kvs = kvsTemp.map(line => line._2)                 //第一部分是null为key，第二部分才是所需数据，为string类型 //根据需求对流进来的数据进行清洗、转换等处理
    val data = kvs.map(_.split("\\t")).filter(_(53) == "finance").map(_(27)).map(_.split("\\?")(0)).filter(! _.contains("iframe")).map((_, 1)) //滑动窗口长度为1小时，滑动间隔为10分钟，这会得到过去1小时内，url和pv的对应关系 //val pvWindow = data.reduceByKeyAndWindow((v1: Int, v2: Int) => v1+v2, Minutes(60), Minutes(10)) //滑动窗口长度为1小时，滑动间隔为10分钟，这同样会得到过去1小时内，url和pv的对应关系，只不过这是加新减旧，第一个参数加上新的，第2个参数，减去上一个batch的。
//和上一个版本的reduceByKeyAndWindow每次都会重新算相比（叠加方式），这种方式（增量方式）更加高效优雅
    val pvWindow = data.reduceByKeyAndWindow(_ + _, _ - _, Minutes(60), Minutes(10)) pvWindow.print() ssc.start() // Start the computation
    ssc.awaitTermination()  // Wait for the computation to terminat
    ssc.stop(true, true)    //优雅地结束
 } }

countByValueAndWindow

countByValueAndWindow的源码如下所示：

 /** * Return a new DStream in which each RDD contains the count of distinct elements in * RDDs in a sliding window over this DStream. Hash partitioning is used to generate * the RDDs with `numPartitions` partitions (Spark's default number of partitions if * `numPartitions` not specified). * @param windowDuration width of the window; must be a multiple of this DStream's * batching interval * @param slideDuration sliding interval of the window (i.e., the interval after which * the new DStream will generate RDDs); must be a multiple of this * DStream's batching interval * @param numPartitions number of partitions of each RDD in the new DStream. */ def countByValueAndWindow( windowDuration: Duration, slideDuration: Duration, numPartitions: Int = ssc.sc.defaultParallelism) (implicit ord: Ordering[T] = null) : DStream[(T, Long)] = ssc.withScope { this.map((_, 1L)).reduceByKeyAndWindow( (x: Long, y: Long) => x + y, (x: Long, y: Long) => x - y, windowDuration, slideDuration, numPartitions, (x: (T, Long)) => x._2 != 0L ) }

reduceByWindow

reduceByWindow的源码如下所示：

/** * Return a new DStream in which each RDD has a single element generated by reducing all * elements in a sliding window over this DStream. However, the reduction is done incrementally * using the old window's reduced value : * 1. reduce the new values that entered the window (e.g., adding new counts) * 2. "inverse reduce" the old values that left the window (e.g., subtracting old counts) * This is more efficient than reduceByWindow without "inverse reduce" function. * However, it is applicable to only "invertible reduce functions". * @param reduceFunc associative and commutative reduce function * @param invReduceFunc inverse reduce function; such that for all y, invertible x: * `invReduceFunc(reduceFunc(x, y), x) = y` * @param windowDuration width of the window; must be a multiple of this DStream's * batching interval * @param slideDuration sliding interval of the window (i.e., the interval after which * the new DStream will generate RDDs); must be a multiple of this * DStream's batching interval */ def reduceByWindow( reduceFunc: (T, T) => T, invReduceFunc: (T, T) => T, windowDuration: Duration, slideDuration: Duration ): DStream[T] = ssc.withScope { this.map((1, _)) .reduceByKeyAndWindow(reduceFunc, invReduceFunc, windowDuration, slideDuration, 1) .map(_._2) }

countByWindow

countByWindow的源码如下所示：

 /** * Return a new DStream in which each RDD has a single element generated by counting the number * of elements in a sliding window over this DStream. Hash partitioning is used to generate * the RDDs with Spark's default number of partitions. * @param windowDuration width of the window; must be a multiple of this DStream's * batching interval * @param slideDuration sliding interval of the window (i.e., the interval after which * the new DStream will generate RDDs); must be a multiple of this * DStream's batching interval */ def countByWindow( windowDuration: Duration, slideDuration: Duration): DStream[Long] = ssc.withScope { this.map(_ => 1L).reduceByWindow(_ + _, _ - _, windowDuration, slideDuration) }

由此可见，countByValueAndWindow、reduceByWindow、countByWindow的底层实现都是“加新减旧”版本的reduceByKeyAndWindow。

上面，求出了每一小时窗口内的Url和Pv的对应关系，如果想求出相同的Url在上一个窗口的Pv和本次窗口的Pv的比值，那么这时侯updateStateByKey，mapWithState就粉墨登场了。由于updateStateByKey和mapWithState二者之间有10倍左右的性能差异。

这里，只涉及mapWithState。

mapWithState

import kafka.serializer.StringDecoder import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.SQLContext import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.streaming._ import org.apache.spark.{SparkContext, SparkConf} object ClickStream { def main (args: Array[String]){ // 屏蔽不必要的日志显示在终端上
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) //创建SparkConf对象，设置应用程序的名称，在程序运行的监控界面可以看到名称
    val conf = new SparkConf().setAppName("ClickStream").setMaster("local[*]") val sc = new SparkContext(conf) //此处设置Batch Interval是在Spark Streaming中生成基本Job的时间单位，窗口和滑动时间间隔一定是该Batch Interval的整数倍
    val ssc = new StreamingContext(sc, Seconds(args(0).toLong)) //由于用到了窗口函数，需要复用前面的RDD，必须checkpoint，注意复用的RDD之间是没有任何关系的
    ssc.checkpoint(args(1)) val topics = Set("clickstream")    //所要获取数据在kafka上的主题
    val brokers = yz4207.hadoop.data.sina.com.cn:19092,yz4203.hadoop.data.sina.com.cn:19092,yz4202.hadoop.data.sina.com.cn:19092,10.39.4.212:19092,10.39.4.201:19092" val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers) //val offset = "largest" //values: smallest, largest ，控制读取最新的数据，还是旧的数据, 默认值为largest //从Spark1.3开始，我们能够使用如下方式高效地从kafka上获取数据
    val kvsTemp = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) val kvs = kvsTemp.map(line => line._2)                 //第一部分是null为key，第二部分才是所需数据，为string类型 //根据需求对流进来的数据进行清洗、转换等处理
    val data = kvs.map(_.split("\\t")).filter(_(53) == "finance").map(_(27)).map(_.split("\\?")(0)).filter(! _.contains("iframe")).map((_, 1)) //滑动窗口长度为1小时，滑动间隔为10分钟，这会得到过去1小时内，url和pv的对应关系 //val pvWindow = data.reduceByKeyAndWindow((v1: Int, v2: Int) => v1+v2, Minutes(60), Minutes(10)) //滑动窗口长度为1小时，滑动间隔为10分钟，这同样会得到过去1小时内，url和pv的对应关系，只不过这是加新减旧，第一个参数加上新的，第2个参数，减去上一个batch的。和上一个版本的reduceByKeyAndWindow每次都会重新算相比（叠加方式），
 //这种方式（增量方式）更加高效优雅
    val pvWindow = data.reduceByKeyAndWindow(_ + _, _ - _, Minutes(60), Minutes(10)) //key是K, value是新值，state是原始值(本batch之前的状态值)。这里你需要把state更新为新值
    val mappingFunc = (key: String, value: Option[Int], state: State[Int]) => { val currentPV = value.getOrElse(0) val output = (key, currentPV, state.getOption().getOrElse(0)) state.update(currentPV) output } //StateSpec只是一个包裹，实际操作仍然是定义的mappingFunc函数
    val urlPvs = pvWindow.mapWithState(StateSpec.function(mappingFunc))    //url,当前batch的PV,上一个batch的PV
 urlPvs.print() ssc.start() // Start the computation
    ssc.awaitTermination()  // Wait for the computation to terminat
    ssc.stop(true, true)    //优雅地结束
 } }