SparkStreaming读取Socket数据

SparkStreaming是7*24h不间断运行，底层操作的是DStream。

读取Socket数据过程

1.SparkStreaming启动后，首先启动一个job，这个job有一个task一直接收数据。
2.task每隔一段时间（batchInterval）就把接收来的数据封装到batch中。生成的每个batch又被封装到一个RDD中，这个RDD又被封装到DStream中。SparkStreaming底层操作的就是DStream
3.DStream有自己的Transformation算子，懒执行需要DStream的outputOperator类算子来触发执行。
4.生成DStream之后，还需启动Job处理DStream

注意：

1.batchInterval表示接收数据的延迟度，可以设置。
2.(集群接收一批次数据的时间 > 处理时间)假如batchInterval=5s，生成一个DStream的时间5s，集群处理一批数据的时间=3s，0-5s集群一直接收数据，5-8s一边接受数据、一边处理数据，8~10s只接受数据…这样每一批次都会造成集群“休息”2s，集群资源不能充分利用。
3.（集群接收一批次数据的时间 < 处理时间）如果集群处理一批次的时间是8s，0-5s一边接受一边处理，10-13s一边接受、一边处理…这样批次数据会越堆越多。如果接收来的数据放内存，会OOM。如果内存不足放Disk，也会加大数据处理的延迟度。
4.最好状态：batchInterval=5s，集群处理一批数据=5s。

读取Socket数据的代码

1.需要设置local[2]，因为一个线程是读取数据，一个线程是处理数据
2.创建StreamingContext有两种方式：val ssc = new StreamingContext(SparkConf ,Durations.seconds(5)) / val ssc = new StreamingContext(SparkContext,Durations.seconds(5))
3.Durations 批次间隔时间的设置需要根据集群的资源情况以及监控每一个job的执行时间来调节出最佳时间
4.SparkStreaming所有业务处理完成之后需要有一个output operato操作
5.StreamingContext.start()straming框架启动之后是不能再次添加业务逻辑
6.StreamingContext.stop()无参的stop方法会将sparkContext一同关闭，stop(false) ,默认为true，会一同关闭
7.StreamingContext.stop()停止之后是不能在调用start

val conf = new SparkConf()
conf.setMaster("local[2]") //1个task接收，1个task处理数据
conf.setAppName("test")
val ssc = new StreamingContext(conf,Durations.seconds(5))val lines = ssc.socketTextStream("node4",9999)
val words = lines.flatMap(one => {one.split(" ")
})
val pairWords = words.map(one=>{(one,1)})
val result = pairWords.reduceByKey(_+_)
result.print(100)ssc.start()
ssc.awaitTermination()ssc.stop()