12.1 SparkStreaming,WordCount

缺乏、安全感 2022-05-16 10:49 169阅读 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

发表评论取消回复

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

相关阅读

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

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

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

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

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

相关 NYOJ 121 另类乘法

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

相关 jenkins 2.121.1 部署项目

相关 LeetCode 121

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