spark streaming 不同于sotm,是一种准实时处理系统。storm 中,把批处理看错是时间教程的实时处理。而在spark streaming中,则反过来,把实时处理看作为时间极小的批处理。
1、三个时间参数
spark streaming 中有三个关于时间的参数,分别如下:
窗口时间windowDuration?:当前窗口要统计多长时间的数据,是批量时间的整数倍
滑动时间slideDuration?:要多长时间更新一次结果,是批量时间的整数倍
批量时间batchDuration?:多长时间创建一个批次,与实际业务无关,只与数据量有关,数据量大则可以设置短一些,数据量小则设置长一些,但必须小于其他两个时间,
2、该怎么设置?
为方便理解,就拿咱们最常见的日启、日活、周启、周活作为示例
注:1、实际中日启、日活、周启、周活更多是用批处理,此处只是拿来方便大家理解
2、此处不是严格意义上的日启、周启。此处的日:最近24小时,周:最近7天
案例1:每隔一小时,统计产品的日启、日活,
窗口时间:1日,滑动时间:1小时,批量时间:1小时、半小时、15分钟、10分钟、5分钟、2分钟均可,视数据量大小而定
案例2:每天统计最近七天累计启动、活跃
窗口时间:7日,滑动时间:1日 批量时间:一小时、半小时、10分钟、5分钟
3、实战
为了理解上边参数是怎么设置的,我们对假定现在有个需求,需要对输入的字母进行计数。
使用nc -lk 9999 模拟生产者,发送数据,streaming 通过socket接收数据
实战1:每10秒统计当前输入的字符
适用:彻底非累加业务
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext} import org.apache.spark.SparkConf val sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]") //10秒创建一个批次 val ssc = new StreamingContext(sparkConf, Seconds(10)) val lines = ssc.socketTextStream("localhost", 9999) val wordCounts = lines.flatMap(_.split(" ")).map(x => (x, 1l)) val current_stream = wordCounts.reduceByKey(_ + _) current_stream.print() current_stream.repartition(1).saveAsTextFiles("/data/socket_wordcount_current.") ssc.start() ssc.awaitTermination()
启动生产者 nc -lk 9999
在spark-shell中输入上边代码
在nc 的终端下,
输入字符操作1、第一个10秒,输入a,第二个10秒输入b,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件,期望结果 【 (a,1) (b,1) 】 原因:我们当前仅输入了a、b
输入字符操作2、第四个10秒,输入c,第五个10秒输入d,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件 期望 【 (c,1) (d,1)】 原因:我们当前输入了c、d
输入字符操作3、这时,不需要操作,等待30秒,在spark-shell中确认第三次计算完成后,查看新产生文件 期望 【 】 原因:当前我们没有输入, 所以没有任何字符可以统计
实战2、每10秒统计历史所有输入的字符。
适用范围:计算历史(包含窗口之外)累计数据,经常用于统计“总装机量”之类
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}
import org.apache.spark.SparkConf
val sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]") //10秒创建一个批次 val ssc = new StreamingContext(sparkConf, Seconds(10)) //累加所有经过的数据 val updateFunc = (values: Seq[Long], state: Option[Long]) => { val currentCount = values.foldLeft(0l)(_ + _) val previousCount = state.getOrElse(0l) Some(currentCount + previousCount) } ssc.checkpoint("socket_wordcount_history") val lines = ssc.socketTextStream("localhost", 9999) val wordCounts = lines.flatMap(_.split(" ")).map(x => (x, 1l)) val history_stream = wordCounts.updateStateByKey[Long](updateFunc)//合并当前数据和历史数据 history_stream.print() history_stream.repartition(1).saveAsTextFiles("/data/socket_wordcount_history.") ssc.start() ssc.awaitTermination()
启动生产者 nc -lk 9999
在spark-shell中输入上边代码
在nc 的终端下,
输入字符操作1、第一个10秒,输入a,第二个10秒输入b,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件,期望结果 【 (a,1) (b,1) 】 原因:我们当前输入了a、b
输入字符操作2、第四个10秒,输入c,第五个10秒输入d,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件 期望 【 (a,1) (b,1) (c,1) (d,1)】 原因:我们当前输入了c、d,历史输入过 a、b
输入字符操作3、这时,不需要操作,等待30秒,在spark-shell中确认第三次计算完成后,查看新产生文件 期望 【 (a,1) (b,1) (c,1) (d,1)】 原因:当前我们没有输入,但是,历史曾经输入过a、b、c、d
输入字符操作4、这时,仍不需要操作,等待30秒,在spark-shell中确认第四次计算完成后,查看新产生文件 期望 【 (a,1) (b,1) (c,1) (d,1)】原因: 当前我们没有输入,但是,历史曾经输入过a、b、c、d
之后,即使没有输入abcd,统计结果仍包含abcd这四个字符各1次
实战3、每隔30秒,统计最近1分钟输入的字母。窗口内历史累加
(适用范围:非累加业务,这里的累加指的是超出window范围)
sc.stop import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}import org.apache.spark.SparkConf val updateFunc = (values: Seq[Long], state: Option[Long]) => { val currentCount = values.foldLeft(0l)(_ + _) val previousCount = state.getOrElse(0l) Some(currentCount + previousCount) } val sparkConf = new SparkConf().setAppName("socket-streaming-wordcount").setMaster("local[2]") val ssc = new StreamingContext(sparkConf, Seconds(10)) //10秒创建一个批次 ssc.checkpoint("socket-kafka-wordcount_recent") val lines = ssc.socketTextStream("localhost", 9999) val words = lines.flatMap(_.split(" ")) val wordCounts = words.map(x => (x, 1l)) val stateDstream = wordCounts.reduceByKeyAndWindow(_ + _, _ - _, Minutes(1), Seconds(30)) //每30秒算一次,数据范围为最近一分钟内收到的数据 另外,使用window时,需要设置checkpoint stateDstream.print() stateDstream.repartition(1).saveAsTextFiles("/data/socket-streaming-wordcount.log") ssc.start() ssc.awaitTermination()
启动生产者 nc -lk 9999
在spark-shell中输入上边代码
在nc 的终端下,
输入字符操作1、第一个10秒,输入a,第二个10秒输入b,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件,期望结果 【 (a,1) (b,1) 】 原因:最近1分钟,我们只输入了a、b
输入字符操作2、第四个10秒,输入c,第五个10秒输入d,等待10秒,在spark-shell中确认第二次计算完成后,查看新产生文件 期望 【 (a,1) (b,1) (c,1) (d,1)】 原因:最近1分钟,我们只输入了a、b、c、d
输入字符操作3、这时,不需要操作,等待30秒,在spark-shell中确认第三次计算完成后,查看新产生文件 期望 【 (a,0) (b,0) (c,1) (d,1)】 原因:最近1分钟,我们只输入了c、d ,1分钟之前输入的a、b将不再在统计范围之内
输入字符操作4、这时,仍不需要操作,等待30秒,在spark-shell中确认第四次计算完成后,查看新产生文件 期望 【 (a,0) (b,0) (c,0) (d,0)】原因:最近1分钟,我们没有任何输入,1 分钟之前输入的a、b、c、d将不再在统计范围之内