imf大数据：第95课：streaming实战，实现在线热搜索词

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

object  class_95_OnlineHottestItems{

/**
* 使用Scala开发集群运行的Spark 在线黑名单过滤程序
* @author DT大数据梦工厂
* 新浪微博：http://weibo.com/ilovepains/
* 不是在线处理数据,都是没有价值的
*
*
*/
def main(args: Array[String]){

val conf=new SparkConf().setMaster("local[2]").setAppName("HottestItems")
val ssc=new StreamingContext(conf,Duration(5))

val soDstream=ssc.socketTextStream("master",9999)

soDstream
.map(_.split(" ")(1))//按照空格分割取出第二个元素
.map(item=>(item,1))//元素转化成turple2
.reduceByKeyAndWindow((v1:Int,v2:Int)=>v1+v2,Seconds(60),Seconds(20))//以60s窗口长度,20s滑动间隔进行元素的累加
.transform(ItemRDD=>ItemRDD//DStream没有排序的算子,所以要转化为rdd
.map(pair=>(pair._2,pair._1))//将k,v对调
.sortByKey(false))//对v进行排序
.map(pair=>(pair._2,pair._1))//将K,V复原
.print()
ssc.start()
ssc.awaitTermination()
}
}