4.spark-streaming实时计算框架三

1.从一台服务器上面接受一行或者多行文本

nc -l 8888

将热度最高的10个网页更新到热门博文板块

在mysql中设计一张表

create table top_web_page(ranking int,htmlID varchar(30),pageHeat double)
create table  table_name(key keyType,key1 keyType)

1.采集数据

2.创建StreamingContext对象并监控streaming/data日志文件目录

val     sc=new SparkConf().setAppName("pagehot")
val     ssc=new StreamingContext(sc,Seconds(5))
val     lines=ssc.textFileStream("home/wang/streaming/data")

val sc=new SparkConf().setAppName();
val ssc=new StreamingContext(sc.Seconds(5))
val lines=ssc.textFileStream()

3.计算网页热度

分隔数据，生成以网页ID为键，热度为值的键值对数据

val html=lines.map{line=>val words=line.split(",");(words(0),0.1*words(1).toInt+0.9*words(2).toInt+0.4*words(3).toDouble+words(4).toInt}
计算每个网页的热度综合
val htmlCount=html.reduceByKeyAndWindow((v1:Double,v2:Double)=>v1+v2,Seconds(60),Seconds(10))

4.按照网页的热度总和降序排序 转换为RDD

val hottestHtml=htmlCount.transform(rdd=>{

    val top10=rdd.sortBy(_._2,false).take(10);
    sc.makeRDD(top10);

})

val hottestHtml=htmlCount.transform(rdd=>{
val top10=rdd.sortBy(_._2,false).take(10);
    sc.makeRDD(top10);

})

启动Spark Streaming