12.1 SparkStreaming,WordCount

缺乏、安全感 2022-05-16 10:49 260阅读 0赞

Storm和SparkStreaming:

1，Storm纯实时的流式处理框架（来一条数据立马进行处理），吞吐量不高

2，SparkStreaming是一个批处理，吞吐量高

3，Storm的事物的机制要比SparkStreaming完善

事物机制是指：接受或者拉取的每一条数据恰好被处理一次

4，SparkStreaming适合做复杂数据处理，Storm只适合做比较简单的汇总型计算

SparkStreaming基于DStream来编程，从DStream里面可以将RDD抽取出来，RDD->DataFrame

5，Storm支持动态资源分配，SparkStreaming是粗粒度的资源调度

Sprark1.6.0开始支持动态调整资源(直接kill掉Excutor,容易丢失数据)

--------------------

WordCountOnline例子--无状态的，意思就是下一次统计出来，上一次的就丢失了：(spark 远程调试)

1，要运行下面的例子，首先打开hadoop1服务器

2，nc -lk 9999在hadoop1服务器上开启一个socket Server客户端，没有nc命令可以yum -y install nc安装

3，启动下面代码，可以再客户端输入了

public class WordCountOnline \{

public static void main(String\[\] args) \{

//一个线程监控hadoop1:9999端口，一个线程执行任务Job：local\[2\]

SparkConf conf = new SparkConf().setMaster("local\[2\]").setAppName("WordCountOnline");

/\*\*

\* 在创建streaminContext的时候 设置batch Interval 设置5秒钟执行一个批次

\*/

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

//监控hadoop1的9999端口

JavaReceiverInputDStream<String> lines = jsc.socketTextStream("hadoop1", 9999);

//接收的数据按规则切割

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() \{

private static final long serialVersionUID = 1L;

@Override

public 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;

@Override

public 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;

@Override

public Integer call(Integer i1, Integer i2) \{

return i1 + i2;

\});

//outputoperator类的算子

counts.print();//输出

jsc.start();

jsc.awaitTermination();

jsc.stop(false);

--------------------

WordCountOnline有状态的显示 --下次显示的时候，以前的数据直接累加

/\*\*

\* UpdateStateByKey的主要功能:

\* 1、Spark Streaming中为每一个Key维护一份state状态，state类型可以是任意类型的的， 可以是一个自定义的对象，那么更新函数也可以是自定义的。

\* 2、通过更新函数对该key的状态不断更新，对于每个新的batch而言，Spark Streaming会在使用updateStateByKey的时候为已经存在的key进行state的状态更新

\* （对于每个新出现的key，会同样的执行state的更新函数操作），

\* hello,3

\* word,2

\* 如果要不断的更新每个key的state，就一定涉及到了状态的保存和容错，这个时候就需要开启checkpoint机制和功能

\* 全面的广告点击分析

\* @author zfg

\* 有何用？ 统计广告点击流量，统计这一天的车流量，统计。。。。点击量

\*/

public class UpdateStateByKeyOperator \{

public static void main(String\[\] args) \{

SparkConf conf = new SparkConf().setMaster("local\[2\]").setAppName("UpdateStateByKeyDemo");

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

/\*\*

\* 多久会将内存中的数据（每一个key所对应的状态）写入到磁盘上一份呢？

\* 如果你的batch interval小于10s 那么10s会将内存中的数据写入到磁盘一份

\* 如果bacth interval 大于10s，那么就以bacth interval为准

\*/

jsc.checkpoint("hdfs://hadoop1:9000/sscheckpoint01");

JavaReceiverInputDStream<String> lines = jsc.socketTextStream("hadoop1", 9999);

JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() \{

private static final long serialVersionUID = 1L;

@Override

public 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;

@Override

public Tuple2<String, Integer> call(String s) \{

return new Tuple2<String, Integer>(s, 1);

\});

//传入参数，

JavaPairDStream<String, Integer> counts = ones.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() \{

private static final long serialVersionUID = 1L;

@Override

public Optional<Integer> call(List<Integer> values, Optional<Integer> state) throws Exception \{

/\*\*

\* values:经过分组最后 这个key所对应的value \[1,1,1,1,1\]

\* state:这个key在本次之前之前的状态

\*/

Integer updateValue = 0 ;

if(state.isPresent())\{

updateValue = state.get();

for (Integer value : values) \{

updateValue += value;

return Optional.of(updateValue);

\});

//output operator

counts.print();

jsc.start();

jsc.awaitTermination();

jsc.close();

--------------------

一般工作当中使用reduceByKeyAndWindow比updateStateByKey多；

在com.java.spark.streaming.WindowOperator里

12.1 SparkStreaming,WordCount

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 121.买卖股票的最佳时机

相关 121. 买卖股票的最佳时机

相关 121.买卖股票最佳时机

相关 bupt|121. 最值问题

相关【框架】121：spring框架之IOC容器

相关 121. 买卖股票的最佳时机

相关 NYOJ 121 另类乘法

相关 jenkins 2.121.1 部署项目

相关 LeetCode 121

相关 figure(figsize=(m,n)) and subplot(121)