[大数据]Sparkstreaming

一.今日小故事

毛遂自荐:平原君准备前往楚国游说楚王联合需要几个门徒,毛遂自己推荐自己,平原君同意了.到了楚国,本来一个早上的时间就可以谈拢的,结果到中午都没有结束,毛遂这个就坐不住了,直接站起来说,直接惹怒楚王,但是毛遂呢,也不害怕他,站起来简明了要害关系,最后楚王同意了联合.

二.springStreaming

1.概念说明:

是一个微批处理的实时计算的框架

原理:是把输入数据以某一时间间隔批量的处理，当批处理间隔缩短到秒级时，即可用于处理实时数据流

2.特点特征:

可以和spark core、sparksql等无缝集成,

支持从多种数据源获取数据, 包括kafka,Flume,Twitter、ZeroMQ、Kinesis以及TCP sockets,然后使用诸如map,reduce,join等高级函数进行复杂的算法运算,最后存储到hdfs文件系统,数据库等;

3.应用场景

无状态操作:只关注当前的DStream中的实时数据，例如 只对当前DStream中的数据做正确性校验

状态操作:对有状态的DStream进行操作时,需要依赖之前的数据 例如 统计网站各个模块总的访问量

window操作:对指定时间段范围内的DStream数据进行操作，例如 需要统计一天之内网站各个模块的访问数量

4.代码实现

//scala里用SparkStreaming实现WordCount的流程
object SparkStreamingTest {
? def main(args: Array[String]): Unit = {
??? //??? 要初始化Spark Streaming程序，必须创建一个StreamingContext对象，
??? //它是所有Spark Streaming功能的主要入口点。
??? //一切都从SparkConf开始
???val conf = new SparkConf().setMaster("local[2]")
.setAppName("NetworkWordCount")
??? //指定时间间隔的ssc初始化
??? val ssc = new StreamingContext(conf, Seconds(1))
???
??? //ssc指定来自TCP源作为输入数据源，即链接一个指定主机的已打开的TCP端口，从该端口中读取文本数据，每行以”\n”作为每行的结尾。
??? val lines = ssc.socketTextStream("localhost", 9999)
???
??? //将DStream进行打平处理,实际是对其内部的离散的rdd进行打平处理
??? val words = lines.flatMap(_.split("\\s+"))
???
??? // 将单列的word转化为双列的kv结构，用于后边的wc操作
??? val pairs = words.map(word => (word, 1))
???
??? //对kv的输出进行wc计算
??? val wordCounts = pairs.reduceByKey(_ + _)
??? //打印wc结果到控制台
??? wordCounts.print()
???
??? //正式开始计算
??? ssc.start()
??? //等待计算结束，一般流式计算没有异常或人为干预是一直保持运行状态的
??? ssc.awaitTermination()
? }
}

流程的抽象说明

1.初始化StreamingContext

2.通过创建DStreams来定义数据源

3.通过将转换和输出操作应用于DStream来定义流式计算

4.开始接收数据并使用它进行处理streamingContext.start()。

5.等待处理停止（手动或由于任何错误）使用streamingContext.awaitTermination()。

6.可以使用手动停止处理streamingContext.stop()。

三.SparkStreaming常见问题说明

1.输入DStream和Receivers

输入DStream:即为从数据流源接收的输入数据流的DStream.

内置两类流媒体源:

基本流数据源:hdfs文件,Socket套接字链接,Akka actor等

高级流数据源:包括Kafka,Flume,Kinesis,ZeroMq等数据源

Receivers:每个输入DStream(除文件流之外)均与Receiver相对关联，该对象负责从流源接收数据并将其存储在Spark的内存中进行处理.

2.Dstreams的输出操作

1).oprint():在运行流应用程序的驱动程序节点上打印DStream中每批数据的前十个元素。???

2).osaveAsTextFiles:将此DStream的内容保存为文本文件

3).osaveAsObjectFiles:将此DStream的内容保存为SequenceFiles序列化Java对象。

4).osaveAsHadoopFiles:将此DStream的内容保存为Hadoop文件。

5).oforeachRDD:最通用的输出运算符，它将函数func应用于从流生成的每个RDD。此函数应将每个RDD中的数据推送到外部系统，例如将RDD保存到文件，或通过网络将其写入数据库。请注意，函数func在运行流应用程序的驱动程序Driver进程中执行。

6).foreach和foreachPartition运行在Worker节点

3.关于SparkStreaming在集群运行时CPU逻辑核心数设置

一个逻辑CPU的资源相当于可以开启一个线程的能力,只有一个的话它就只有一个线程来处理工作,而DStream必须得有一个线程用来接受数据,就没有多余的线程处理数据. 所以需要最少2个cpu用来执行,也就是两个线程