查看: 337|回复: 8

【转】【Spark篇】---SparkStream初始与应用

[复制链接]
论坛徽章:
20
Mycat徽章
日期:2016-06-08 14:19:56架构徽章
日期:2018-01-11 17:24:29Hadoop研习者初级
日期:2018-01-11 17:33:01架构徽章
日期:2018-02-08 16:19:13Go徽章
日期:2018-05-10 16:26:29postgresql徽章
日期:2018-05-10 16:26:58股票徽章
日期:2018-06-29 11:14:50股票徽章
日期:2018-08-30 15:33:52股票徽章
日期:2018-11-22 15:36:24Java徽章
日期:2018-01-04 17:15:14bash徽章
日期:2017-12-15 16:07:12Oracle研习者初级
日期:2016-06-16 10:32:31
发表于 2019-5-1 23:41 | 显示全部楼层 |阅读模式

tm HTML Spark

【Spark篇】---SparkStream初始与应用https://www.cnblogs.com/LHWorldBlog/p/8435423.html


回复

使用道具 举报

论坛徽章:
20
Mycat徽章
日期:2016-06-08 14:19:56架构徽章
日期:2018-01-11 17:24:29Hadoop研习者初级
日期:2018-01-11 17:33:01架构徽章
日期:2018-02-08 16:19:13Go徽章
日期:2018-05-10 16:26:29postgresql徽章
日期:2018-05-10 16:26:58股票徽章
日期:2018-06-29 11:14:50股票徽章
日期:2018-08-30 15:33:52股票徽章
日期:2018-11-22 15:36:24Java徽章
日期:2018-01-04 17:15:14bash徽章
日期:2017-12-15 16:07:12Oracle研习者初级
日期:2016-06-16 10:32:31
 楼主| 发表于 2019-5-1 23:41 | 显示全部楼层
一、前述  SparkStreaming是流式处理框架,是Spark API的扩展,支持可扩展、高吞吐量、容错的实时数据流处理,实时数据的来源可以是:Kafka, Flume, Twitter, ZeroMQ或者TCP sockets,并且可以使用高级功能的复杂算子来处理流数据。例如:map,reduce,join,window 。最终,处理后的数据可以存放在文件系统,数据库等,方便实时展现。  二、SparkStreaming与Storm的区别  1、Storm是纯实时的流式处理框架,SparkStreaming是准实时的处理框架(微批处理)。因为微批处理,SparkStreaming的吞吐量比Storm要高。  2、Storm 的事务机制要比SparkStreaming的要完善。  3、Storm支持动态资源调度。(spark1.2开始和之后也支持)  4、SparkStreaming擅长复杂的业务处理,Storm不擅长复杂的业务处理,擅长简单的汇总型计算。
回复 支持 反对

使用道具 举报

论坛徽章:
20
Mycat徽章
日期:2016-06-08 14:19:56架构徽章
日期:2018-01-11 17:24:29Hadoop研习者初级
日期:2018-01-11 17:33:01架构徽章
日期:2018-02-08 16:19:13Go徽章
日期:2018-05-10 16:26:29postgresql徽章
日期:2018-05-10 16:26:58股票徽章
日期:2018-06-29 11:14:50股票徽章
日期:2018-08-30 15:33:52股票徽章
日期:2018-11-22 15:36:24Java徽章
日期:2018-01-04 17:15:14bash徽章
日期:2017-12-15 16:07:12Oracle研习者初级
日期:2016-06-16 10:32:31
 楼主| 发表于 2019-5-1 23:41 | 显示全部楼层
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。
回复 支持 反对

使用道具 举报

论坛徽章:
20
Mycat徽章
日期:2016-06-08 14:19:56架构徽章
日期:2018-01-11 17:24:29Hadoop研习者初级
日期:2018-01-11 17:33:01架构徽章
日期:2018-02-08 16:19:13Go徽章
日期:2018-05-10 16:26:29postgresql徽章
日期:2018-05-10 16:26:58股票徽章
日期:2018-06-29 11:14:50股票徽章
日期:2018-08-30 15:33:52股票徽章
日期:2018-11-22 15:36:24Java徽章
日期:2018-01-04 17:15:14bash徽章
日期:2017-12-15 16:07:12Oracle研习者初级
日期:2016-06-16 10:32:31
 楼主| 发表于 2019-5-1 23:42 | 显示全部楼层

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

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

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

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

使用道具 举报

新浪微博达人勋 xenron  实名认证
论坛徽章:
227
matlab徽章
日期:2019-07-11 14:23:31股票徽章
日期:2019-07-11 14:24:09智慧眼徽章
日期:2019-07-11 14:24:43股票徽章
日期:2019-07-11 14:24:48
发表于 2019-5-4 06:09 | 显示全部楼层
receiver模式下接受数据,local的模拟线程必须大于等于2,一个线程用来receiver用来接受数据,另一个线程用来执行job。
Durations时间设置就是我们能接收的延迟度。这个需要根据集群的资源情况以及任务的执行情况来调节。
回复 支持 反对

使用道具 举报

新浪微博达人勋 xenron  实名认证
论坛徽章:
227
matlab徽章
日期:2019-07-11 14:23:31股票徽章
日期:2019-07-11 14:24:09智慧眼徽章
日期:2019-07-11 14:24:43股票徽章
日期:2019-07-11 14:24:48
发表于 2019-5-4 06:10 | 显示全部楼层
所有的代码逻辑完成后要有一个output operation类算子触发执行。
JavaStreamingContext.start() Streaming框架启动后不能再次添加业务逻辑。
JavaStreamingContext.stop() 无参的stop方法将SparkContext一同关闭,stop(false),不会关闭SparkContext,方便后面使用
JavaStreamingContext.stop()停止之后不能再调用start。
回复 支持 反对

使用道具 举报

新浪微博达人勋 xenron  实名认证
论坛徽章:
227
matlab徽章
日期:2019-07-11 14:23:31股票徽章
日期:2019-07-11 14:24:09智慧眼徽章
日期:2019-07-11 14:24:43股票徽章
日期:2019-07-11 14:24:48
发表于 2019-5-4 17:15 | 显示全部楼层
local的模拟线程数必须大于等于2 因为一条线程被receiver(接受数据的线程)占用,另外一个线程是job执行
回复 支持 反对

使用道具 举报

新浪微博达人勋 xenron  实名认证
论坛徽章:
227
matlab徽章
日期:2019-07-11 14:23:31股票徽章
日期:2019-07-11 14:24:09智慧眼徽章
日期:2019-07-11 14:24:43股票徽章
日期:2019-07-11 14:24:48
发表于 2019-5-4 17:15 | 显示全部楼层
Durations时间的设置,就是我们能接受的延迟度,这个我们需要根据集群的资源情况以及监控每一个job的执行时间来调节出较佳时间。
回复 支持 反对

使用道具 举报

新浪微博达人勋 xenron  实名认证
论坛徽章:
227
matlab徽章
日期:2019-07-11 14:23:31股票徽章
日期:2019-07-11 14:24:09智慧眼徽章
日期:2019-07-11 14:24:43股票徽章
日期:2019-07-11 14:24:48
发表于 2019-5-4 17:16 | 显示全部楼层
receiver  task是7*24小时一直在执行,一直接受数据,将一段时间内接收来的数据保存到batch中。假设batchInterval为5s,那么会将接收来的数据每隔5秒封装到一个batch中,batch没有分布式计算特性,这一个batch的数据又被封装到一个RDD中,RDD最终封装到一个DStream中。
回复 支持 反对

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册 新浪微博登陆

本版积分规则

 

GMT+8, 2019-7-16 11:57 , Processed in 0.166194 second(s), 45 queries .