码迷,mamicode.com
首页 > 其他好文 > 详细

spark streaming测试之四设置窗口大小接收数据

时间:2015-05-31 18:45:23      阅读:115      评论:0      收藏:0      [点我收藏+]

标签:streaming   package   window   import   程序   

测试思路:

    首先,使用网络数据发送程序发送数据;

    然后,运行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()
  }
}

本文出自 “一步.一步” 博客,转载请与作者联系!

spark streaming测试之四设置窗口大小接收数据

标签:streaming   package   window   import   程序   

原文地址:http://snglw.blog.51cto.com/5832405/1656796

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!