Spark _29_SparkStreaming初始

SparkStreaming简介

SparkStreaming是流式处理框架，是Spark API的扩展，支持可扩展、高吞吐量、容错的实时数据流处理，实时数据的来源可以是：Kafka, Flume, Twitter, ZeroMQ或者TCP sockets，并且可以使用高级功能的复杂算子来处理流数据。例如：map,reduce,join,window 。最终，处理后的数据可以存放在文件系统，数据库等，方便实时展现。

SparkStreaming与Storm的区别

Storm是纯实时的流式处理框架，SparkStreaming是准实时的处理框架（微批处理）。因为微批处理，SparkStreaming的吞吐量比Storm要高。

Storm 的事务机制要比SparkStreaming的要完善。

Storm支持动态资源调度。(spark1.2开始和之后也支持)

SparkStreaming擅长复杂的业务处理，Storm不擅长复杂的业务处理，擅长简单的汇总型计算。

SparkStreaming初始

SparkStreaming初始理解

注意:

receiver task是7*24小时一直在执行，一直接受数据，将一段时间内接收来的数据保存到batch中。假设batchInterval为5s,那么会将接收来的数据每隔5秒封装到一个batch中，batch没有分布式计算特性，这一个batch的数据又被封装到一个RDD中，RDD最终封装到一个DStream中。

例如：假设batchInterval为5秒，每隔5秒通过SparkStreaming将得到一个DStream,在第6秒的时候计算这5秒的数据，假设执行任务的时间是3秒,那么第6~9秒一边在接收数据，一边在计算任务，9~10秒只是在接收数据。然后在第11秒的时候重复上面的操作。

如果job执行的时间大于batchInterval会有什么样的问题？

如果接受过来的数据设置的级别是仅内存，接收来的数据会越堆积越多，最后可能会导致OOM（如果设置StorageLevel包含disk, 则内存存放不下的数据会溢写至disk, 加大延迟）。

SparkStreaming代码

代码注意事项：

启动socket server 服务器：nc –lk 9999
receiver模式下接受数据，local的模拟线程必须大于等于2，一个线程用来receiver用来接受数据，另一个线程用来执行job。
Durations时间设置就是我们能接收的延迟度。这个需要根据集群的资源情况以及任务的执行情况来调节。
创建JavaStreamingContext有两种方式（SparkConf,SparkContext）
所有的代码逻辑完成后要有一个output operation类算子。
JavaStreamingContext.start() Streaming框架启动后不能再次添加业务逻辑。
JavaStreamingContext.stop() 无参的stop方法将SparkContext一同关闭，stop(false)，不会关闭SparkContext。
JavaStreamingContext.stop()停止之后不能再调用start。

javaAPI：

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("WordCountOnline");
/*** 在创建streaminContext的时候 设置batch Interval*/
JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));JavaReceiverInputDStream<String> lines = jsc.socketTextStream("node5", 9999);JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Iterable<String> call(String s) {return Arrays.asList(s.split(" "));}
});JavaPairDStream<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<String, Integer> call(String s) {return new Tuple2<String, Integer>(s, 1);}
});JavaPairDStream<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Integer call(Integer i1, Integer i2) {return i1 + i2;}
});//outputoperator类的算子   counts.print();jsc.start();//等待spark程序被终止jsc.awaitTermination();jsc.stop(false);

scalaAPI:

package com.sparkStreamingimport org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}object StreamTest {def main(args: Array[String]): Unit = {//在这里将local线程设置两个，否则会只能接受数据，无法处理val conf = new SparkConf().setMaster("local[2]").setAppName("st")
//    val sc = new SparkContext(conf)val ssc = new StreamingContext(conf,Durations.seconds(5))
//    new StreamingContext(sc,Durations.seconds(5))ssc.sparkContext.setLogLevel("error")val lines = ssc.socketTextStream("henu1",9999)val words = lines.flatMap(one => {one.split(" ")})val map = words.map(one => {(one,1)})val result = map.reduceByKey(_+_)result.print(100)ssc.start()ssc.awaitTermination()ssc.stop()}
}

【注】启动前先在对应虚拟机启动

[root@henu1 ~]# nc -lk 9999

如果没有nc。下载

[root@henu1 ~]# yum install -y nc

再执行 nc -lk 9999

否则报错：

然后在linux端输入字段：

idea端：

在浏览器输入localhost:4040,看其UI页面：

具体就不多说了。

Spark _29_SparkStreaming初始相关推荐

【核心API开发】Spark入门教程[3]
本教程源于2016年3月出版书籍<Spark原理.机制及应用> ,在此以知识共享为初衷公开部分内容,如有兴趣,请支持正版书籍. Spark综合了前人分布式数据处理架构和语言的优缺点,使用简 ...
【Spark篇】---SparkStream初始与应用
一.前述 SparkStreaming是流式处理框架,是Spark API的扩展,支持可扩展.高吞吐量.容错的实时数据流处理,实时数据的来源可以是:Kafka, Flume, Twitter, Zer ...
【Spark篇】---Spark初始
一.前述 Spark是基于内存的计算框架,性能要优于Mapreduce,可以实现hadoop生态圈中的多个组件,是一个非常优秀的大数据框架,是Apache的顶级项目.One stack rule ...
2021年大数据Spark（五十二）：Structured Streaming 事件时间窗口分析
目录事件时间窗口分析时间概念 event-time 延迟数据处理延迟数据 Watermarking 水位官方案例演示事件 ...
2021年大数据Spark（十五）：Spark Core的RDD常用算子
目录常用算子基本算子分区操作函数算子重分区函数算子 1).增加分区函数 2).减少分区函数 3).调整分区函数聚合函数算子 Scala集合中的聚合函数 ...
spark—3（Spark Scheduler）
2019独角兽企业重金招聘Python工程师标准>>> Spark的核心是根据RDD来实现的,Spark Scheduler则为Spark核心实现的重要一环,其作用就是任务调度.Sp ...
Spark菜鸟学习营Day5 分布式程序开发
Spark菜鸟学习营Day5 分布式程序开发这一章会和我们前面进行的需求分析进行呼应,完成程序的开发. 开发步骤分布式系统开发是一个复杂的过程,对于复杂过程,我们需要分解为简单步骤的组合. 针对每 ...
9.spark core之共享变量
简介 spark执行操作时,可以使用驱动器程序Driver中定义的变量,但有时这种默认的使用方式却并不理想. 集群中运行的每个任务都会连接驱动器获取变量.如果获取的变量比较大,执行效率会非常低下. ...
spark算子_Spark常用算子
Spark的算子分类: 从大方向说,Spark算子大致可以分为以下两类: (1)Transformation变换/转换算子:这种变换并不触发提交作业,这种算子是延迟执行的,也就是说从一个RDD转换生成 ...