ITPub博客

首页 > 大数据 > 数据分析 > spark-streaming之 socketTextStream

spark-streaming之 socketTextStream

原创 数据分析 作者:hgs19921112 时间:2018-10-17 23:10:34 0 删除 编辑
package hgs.spark.streaming
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.HashPartitioner
object SocketStreamingTest {
  def main(args: Array[String]): Unit = {
    
    val conf = new SparkConf();
    conf.setMaster("local[2]").setAppName("SocketStreaming")
    val context = new SparkContext(conf);
    //要添加spark-streaming的依赖包,spark的Seconds
    val streamContext  = new StreamingContext(context,Seconds(5));
    val ds = streamContext.socketTextStream("192.168.6.129", 8888, StorageLevel.MEMORY_ONLY);
    streamContext.checkpoint("d:\\chekpoint")
    //val ds2 = ds.flatMap(_.split(" ")).map((_,1)).reduceByKey((x,y)=>(x+y))//这种方式只是对该批次数据进行处理,并没有累计上一个批次
    
    
    //updateFunc: (Iterator[(K, Seq[V], Option[S])]) K:单词, Seq[V]该批次单词出现次数列表,Option:上一次计算的结果
    val updateFunc=(iter:Iterator[(String,Seq[Int],Option[Int])])=>{
      //iter.flatMap(it=>Some(it._2.sum+it._3.getOrElse(0)).map((it._1,_)))//方式一
      //iter.flatMap{case(x,y,z)=>{Some(y.sum+z.getOrElse(0)).map((x,_))}}//方式二
      iter.flatMap(it=>Some(it._1,(it._2.sum.toInt+it._3.getOrElse(0))))//方式三
    }
    val partitionner = new HashPartitioner(2)
    //通过updateStatByKey来进行累加
    val ds2 = ds.flatMap(_.split(" ")).map((_,1)).updateStateByKey(updateFunc, partitionner, true)
    //打印
    ds2.print()
    streamContext.start()
    streamContext.awaitTermination()
  }
}


来自 “ ITPUB博客 ” ,链接:http://blog.itpub.net/31506529/viewspace-2216732/,如需转载,请注明出处,否则将追究法律责任。

上一篇: day2 Java NIO
下一篇: spark 与flume 1.6.0
请登录后发表评论 登录
全部评论

注册时间:2017-11-22

  • 博文量
    105
  • 访问量
    145774