十年网站开发经验 + 多家企业客户 + 靠谱的建站团队
量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决
测试思路:
专注于为中小企业提供网站制作、网站设计服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业巧家免费做网站提供优质的服务。我们立足成都,凝聚了一批互联网行业人才,有力地推动了近1000家企业的稳健成长,帮助中小企业通过网站建设实现规模扩充和转变。
首先,使用网络数据发送程序发送数据;
然后,运行spark程序;
观察效果。
说明:
1. 这里也需要设置检查点目录
2. 这里有四个参数:
前两个分别是监听的端口和每隔多少毫秒接收一次数据;
第三个参数是接收前多少毫秒的数据;(详细请参见window具体含义)
第四个参数是每隔多少毫秒接收一次数据。
sparkStreaming import org.apache.log4j.{LoggerLevel} import org.apache.spark.storage.StorageLevel import org.apache.spark.streaming.{SecondsStreamingContext} import org.apache.spark.{SparkContextSparkConf} import org.apache.spark.streaming.StreamingContext._ WindowWordCount { def main(args: Array[]){ Logger.().setLevel(Level.) Logger.().setLevel(Level.) conf = SparkConf().setAppName().setMaster() sc = SparkContext(conf) ssc = StreamingContext(sc()) ssc.checkpoint() val lines = ssc.socketTextStream(args(0),args(1).toInt, StorageLevel.MEMORY_ONLY_SER) words = lines.flatMap(_.split()) //windows operator val wordCounts = words.map(x=>(x,1)).reduceByKeyAndWindow((a:Int,b:Int)=>(a+b), Seconds(args(2).toInt),Seconds(args(3).toInt)) wordCounts.print() ssc.start() ssc.awaitTermination() } }