DStream操作实战:2.SparkStreaming接受socket数据,实现单词计数累加
2018-03-11 22:08
393 查看
package cn.testdemo.dstream.socket
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
//todo:利用sparkStreaming接受socket数据,实现所有批次数据单词出现次数的总和
object SparkStreamingSocketTotal {
//currentValues:表示的是当前批次中相同单词出现所有的1 (hadoop,1)(hadoop,1)(hadoop,1)
//historyValue:表示之前所有批次中单词出现的总次数
def updateFunc(currentValues:Seq[Int], historyValue:Option[Int]):Option[Int] ={
val newValues: Int = currentValues.sum + historyValue.getOrElse(0)
Some(newValues)
}
def main(args: Array[String]): Unit = {
//1、创建sparkConf
val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketTotal").setMaster("local[2]")
//2、创建sparkContext
val sc = new SparkContext(sparkConf)
sc.setLogLevel("WARN")
//3、创建streamingContext
val ssc = new StreamingContext(sc,Seconds(5))
//设置checkpoint,用于保存中间结果数据
ssc.checkpoint("./ck2018")
//4、获取socket数据
val stream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.216.121",9999)
//5、操作数据流
val wordAndOne: DStream[(String, Int)] = stream.flatMap(_.split(" ")).map((_,1))
//6、相同单词出现的次数累加---updateStateByKey方法,它会记录下上一个批次key的结果
val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
//7、打印结果数据
result.print()
//8、开启计算
ssc.start()
ssc.awaitTermination()
}
}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
//todo:利用sparkStreaming接受socket数据,实现所有批次数据单词出现次数的总和
object SparkStreamingSocketTotal {
//currentValues:表示的是当前批次中相同单词出现所有的1 (hadoop,1)(hadoop,1)(hadoop,1)
//historyValue:表示之前所有批次中单词出现的总次数
def updateFunc(currentValues:Seq[Int], historyValue:Option[Int]):Option[Int] ={
val newValues: Int = currentValues.sum + historyValue.getOrElse(0)
Some(newValues)
}
def main(args: Array[String]): Unit = {
//1、创建sparkConf
val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketTotal").setMaster("local[2]")
//2、创建sparkContext
val sc = new SparkContext(sparkConf)
sc.setLogLevel("WARN")
//3、创建streamingContext
val ssc = new StreamingContext(sc,Seconds(5))
//设置checkpoint,用于保存中间结果数据
ssc.checkpoint("./ck2018")
//4、获取socket数据
val stream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.216.121",9999)
//5、操作数据流
val wordAndOne: DStream[(String, Int)] = stream.flatMap(_.split(" ")).map((_,1))
//6、相同单词出现的次数累加---updateStateByKey方法,它会记录下上一个批次key的结果
val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
//7、打印结果数据
result.print()
//8、开启计算
ssc.start()
ssc.awaitTermination()
}
}
相关文章推荐
- DStream操作实战:1.SparkStreaming接受socket数据,实现单词计数WordCount
- DStream操作实战:3.SparkStreaming开窗函数reduceByKeyAndWindow,实现单词计数
- 大数据IMF传奇行动绝密课程第103课:动手实战Spark Streaming Broadcast、Accumulator实现在线黑名单过滤和计数
- 用socket发送流数据示--用 php://input? 接受post数据(可实现php和c/c++数据通讯)
- Python Streaming实战2: Join的实现与数据过滤
- imf大数据:第95课:streaming实战,实现在线热搜索词
- 在idea上用SparkStreaming实现从远程socket读取数据并完成Wordcount
- 大数据IMF传奇行动绝密课程第94课:SparkStreaming实现广告计费系统中在线黑名单过滤实战
- 设计模式实战:在WinForm中用Command模式实现可以撤销的数据操作
- Spark Streaming实现实时WordCount,DStream的使用,updateStateByKey(func)实现累计计算单词出现频率
- Android 通过Socket实现手机端向PC发送数据,并接受PC端返回的数据
- 大数据IMF传奇行动绝密课程第95课:通过SparkStreaming的window操作实战模拟新浪微博、百度、京东等热点搜索词案例实战
- hadoop streaming两个数据文件实现join合并操作
- 接受、显示和操作公共数据成员
- Scott Mitchell 的ASP.NET 2.0数据操作教程之十:使用 GridView 和DetailView实现的主/从报表
- 利用反射和自定义特性实现基本数据存取操作自动获取
- 关于socket编程数据发送和接受的感受(wm和pc之间)
- python中socket接受数据的三种方法
- Adodb.Stream是ADO的Stream对象,提供存取二进制数据或者文本流,从而实现对流的读、写和管理等操作.
- ASP.NET 2.0 中的数据操作:: 使用 GridView 和DetailView实现的主/从报表