Spark_总结

朴灿烈づ我的快乐病毒、 2022-04-16 20:38 205阅读 0赞

##### Storm和SparkStreaming区别 #####

<table> 
 <thead> 
  <tr> 
   <th>storm</th> 
   <th>sparkStreaming</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>纯实时的流式处理，来一条数据就立刻进行处理</td> 
   <td>微批处理，每次处理的都是一批非常小的数据</td> 
  </tr> 
 </tbody> 
</table>

storm 支持动态吊证并行度（动态的资源分配），sparkstreaming（粗粒度，比较消耗资源）

##### storm优点 || 缺点 #####

storm 流式计算（扶梯）：

- 优点： 数据延迟度很低，storm的事务机制要比sparkstreaning 的事务机制要完善（什么是事务机制？对于一条数据，不多处理也不少处理，对于一条数据恰好处理一次，比如 金融，股票等要求实时性比较高，就需要选storm）
    - 缺点 ：一致持有资源，每一条数据都要在急群众某一台节点处理，要计算的数据会进行网络传输，吞吐量小，另外storm不适合做复杂的业务逻辑（适合汇总）

SparkStreaming 优缺点：

sparkstreaming 微批处理（类似于电梯） 并不是纯的批处理
    - 优点 :吞吐量大，可以做复杂的业务逻辑（保证每个job的处理小鱼batch interval）
    - 缺点: 数据延迟较高
    公司中为什么选用sparkstreaming 要多一些？
    1. 秒级延迟，通常应用程序是可以接受的
    2. 可以应用机器学习，sparkSQL ..可扩展性比较好，数据吞吐量较高

##### SparkStreaming #####

###### 什么是sparkStreaming？ ######

sparkstreaming 是一个流式处理框架，处理的模式是微批处理（微批多大，通过实践来设置这个批多大，[for example: Batch Interval 5s]）
    sparkStreaming 基于DStream(Discretized Streams:离散的数据流）来进行编程，处理的是一个流，这个流什么时候切成一个rdd->根据batchinterval来决定何时切割成一个RDD

###### sparkstreaming 框架图 ######

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70][]  
job个数是由output operator决定的，streamContext底层封装了SparkContect  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 1]  
从图上可以看到，Batch Interval的间隔是5s，也就是说每经过5s，SparkStreaming会将这5s内的信息封装成一个DStream,然后提交到Spark集群进行计算

##### 执行过程 #####

在这里插入代码片

###### 注意 ######

如果这个job执行的时间大于5s会有什么问题？
    数据在5s内处理不完，会启动另一个job，导致数据越积越多，从而导致SparkStreaming  down

##### SparkStreaming代码TransformOperator #####

###### 案例：过滤黑名单 ######

这里模拟了一份黑名单，SparkStreaming监控服务器中指定端口，时间设定为每5秒处理一次数据。每当job触发时，用户输入的数据与黑名单中的数据进行左外连接，然后过滤

node1 创建一个Socket Server

nc -lk 8888(页面停止，开始输入数据进入8888 端口，此时SparkStreaming监听这个端口)
    	//nc的-l参数表示创建一个监听端口，等待新的连接。-k参数表示当前连接结束后仍然保持监听，必须与-l参数同时使用。
    	hello world
    	hello jack
    	hello tom(过滤tom)

result:  
![在这里插入图片描述][20181113072839523.png]  
注意事项：

1. 为什么没有数据？
    	因为只开启了一条线程（这里只有接受数据的线程），所以local的模拟sparkStreaming必须至少设置两个线程
    	new SparkConf().setMaster("loacal[2]").setAppName(TransformBlacklist");
    2. Durations时间的设置---接收数据的延迟时间，多久触发一次Job
    	final JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
    3. 创建JavaStreamingContext有两种方式 (sparkconf, sparkcontext)
    4. 业务逻辑完成后，需要有一个output operator, 将sparkstreaming 处理后的数据输出(HDFS, DBMS)
    5.  关于JavaStreamingContext的start() 或 stop()
    	 javaStreamingContext.start() //starming 框架启动之后不能再添加业务逻辑
    	 JavaStreamingContext.stop() // 无参的stop方法会将sparkContext一同关闭，解决方法 ：stop(false)
    	 javastreamingContex.stop() //停止之后不能再调用start()
    6. DStreams  （离散的流），应用在每个DStream的算里操作，应用在RDD,应用在Partition,应用在Partition中的一条条数据，所以最终应用在每一条记录上

##### Window窗口操作 #####

window operation

普通的 每隔多长时间切割RDD
     基于窗口的操作：每隔多长时间切割RDD，每隔多长时间计算一次，每次计算的量是多少

为什么需要有窗口操作？  
比如别人要求能够实时看到此刻之前一段时间的数据情况，如果使用批处理的话，那么我们只能固定一个整段时间然后对这个整段时间进行spark core的计算，但是别人的要求是每一个时刻都要有结果，那么久需要窗口操作？ 但是窗口操作肯定会有很多的重复计算，这里有一个优化的地方（这个优化也不是必须的，视情况而定，比如我们要查看最近30分钟最热门的头条，骂我们再设计的时候不可能每隔30分钟计算一次，这里定义了滑动窗口的时间是1分钟，然后计算是30分钟内的数据，那么肯定会有29分钟重复的数据计算）； 但是优化的话就会有一个前提，必须要checkpoint

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 2]  
问题一：batch interval 5s，窗口大小可以是8s吗？  
不行，有的batch就不能被窗口所包含，必须是batch interval的整数倍  
问题二： 滑动窗口时间8s可以么？  
必须是batch interval的整数倍  
优化： 如何避免time3 被重新计算,可以没有，但是有的话，就需要这种优化  
![Batch Interval 1s   ||   窗口大小 5s   ||   滑动窗口 1s][Batch Interval 1s _ _ 5s _ _ 1s]  
思考：计算一个趋势的时候，需要基于华东窗口的操作，是否必须优化，避免重复计算? (未必)  
For example:

1.  查看微博中每小时的热门微博，每隔1分钟计算一次，相当于重复计算了59分钟的内容
2.  商家想看前5分钟的销售额，每隔30秒看一次，也需要基于窗口的操作

##### UpdateStateByKey #####

updateStateBuykey的主要功能

1.  Spark Streaming中为每一个key维护一份state状态，这个state类型可以是任意类型的，可以是一个自定义的对象，那么更新函数也可以是任意类型的。
2.  通过更新函数对可以key的状态不断更新，对于每个新的batch而言，Spark Streaming会使用updateStateByKey的时候为已经存在的key进行state的状态更新
3.  如果要不断的更新每个key的state,就涉及了状态的保存和容错，这个时候就需要开启checkpoint机制和功能

##### UpdateStateByKey 用处:统计广告点击流量，统计一天的车流量… #####

案例： 全国的广告点击分析UpdateStateBykeyoperator  
这里做了checkpoint操作，jsc.checkpoint(“hdfs://ndoe1:8020/user/sscheckpoint”);  
node1创建一个Socket Server，指定8888端口，SparkStreaming与服务器这个端口建立通信，那么用户的数据从这里流向SparkStreaming进行计算。  
在这个案例中，用以空格分割的单词来模拟用户点击的广告，每个单词代表一个广告，统计每一个广告（单词）出现的次数（就是WordCount）  
最后的conunts.print() //output operator类型的算子

result: 利用SparkStreaming做到了微批处理，近似实时计算  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 3]  
查看hdfs,发现设置checkpoint会将SparkStreaming的处理结果进行了持久化  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 5]

![在这里插入图片描述][20181113072921569.png]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 6]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 7]

##### 基于滑动窗口的热点搜索实时统计–WindowOperatorr #####

1.  未优化的
2.  优化的必须设置checkpoint的目录  
    以下是优化过的reduceByKeyAndWindow  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 8]  
    补充：

Spark mater 8080端口 监控资源
    	Driver 4040端口， 监控任务，可以看到有一个Streaming job(他里面有一个线程，是一致运行的，负责接受我们的数据)
    2.transfom 和foreachRDD的区别？
    transform  Transformation类型算子，transform非常厉害，可以拿到每一个DStream的rdd;这样就可以利用每一个rdd的所有算子来转换；甚至在里面应用spark core，或者将rdd转换成DataFrame来使用SparkSQL操作
       foreachRDD Action类型算子，对于每个RDD进行操作，什么时候会用？最后存结果的时候会用

3.transform取出DStream中的RDD  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 9]

使用transform将DStream中的RDD抽取出来，调用了RDD的Action类的算子（是可以执行的）是在Driver端执行的，如果不在Driver端执行，调用Action类的算子就不会触发一个job了
     
        对RDD的操作才会发送到Executor端执行，transform是对DStream进行操作(transform中抽取RDD，对这个RDD执行collect类型的数据，在job Generator时执行的，生成了多个job，以jobSet的形式发送给jobSecheduler)，这样的话就可以预警：对数据的预警，与标准进行比较，如果超过了这个标准就进行报警（一旦发现某个黑名单就立即进行报警，），整体的代码是在Driver端执行的，但是部分代码对RDD的操作是在Executor段执行的
        SparkContext sc = userClickRDD.context();
        Object obj = "可以来源于数据库，动态的更改广播变量"
        sc.broadCast(obj)

##### SparkStreaming–Driver HA #####

2.6.1Driver也有可能挂掉，如何实现它的高可用？
    
     
          当一个Driver挂掉后，（回忆：当初的Master是由zookeeper进行托管），另外启动一个Driver，它就需要从上一个Driver中获得相关的信息(包括batch的进度，data的位置，job执行进度，DStream的Graph(基于DStream的业务逻辑))
     
          如何实现Driver的高可用-->基于HDFS上面的元数据（Driver的信息）进行恢复，注意！不会重新new SparkContext,因为这样相当于又创建了一个全新的Driver
    2.6.2Driver HA的代码套路
        1.指定了去哪一个目录下面寻找Driver的元数据信息
        2.提交Application到集群中执行的时候，必须使用cluster模式，同时必须指定一个参数 --supervise（当某一个Driver挂掉，新的Driver需要另一个Driver中的信息来继续job的执行）
    2.6.3监控HDFS上指定目录下文件数量的变化
    示例代码SparkStreamingOnHDFS
        1.为了状态的保存和容错，开启了checkpoint机制，Driver HA
       2.ssc.textFileStream("hdfs://node1:8020/userhdfs/")   //监控hdfs上/user/hdfs的变化
     
    命令：hadoop fs -put wc /user/hdfs
    2.6.4SparkStreaming 监控 HDFS 上文件数量的变化，并将变化写入到MySql中
    示例代码SparkStreamingOnHDFSToMySQL
       1.为了状态的保存和容错，开启了checkpoint机制，Driver HA
       2.ssc.textFileStream("hdfs://node1:8020/userhdfs/")   //监控hdfs上/user/hdfs的变化

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70]: /images/20220417/0275aadece3448b9b30d3ec8596096a8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 1]: /images/20220417/c4073b7355a040ea8f2cded90916d7a3.png
[20181113072839523.png]: /images/20220417/f4be3536efe342b3874ab5d7065f3da2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 2]: /images/20220417/852721ead3354af8ace0bf343fa8e6c1.png
[Batch Interval 1s _ _ 5s _ _ 1s]: /images/20220417/9b232effb7b64aa0bd5f6a551df5d574.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 3]: /images/20220417/ec4786297aa54e9bb24ca2f2e48aa68a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 4]: /images/20220417/28a09aaf52444954a4929ebd0aecca82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 5]: /images/20220417/e1b61aded9184d85885c3ff946dacbfc.png
[20181113072921569.png]: /images/20220417/8a830a10ab934f3c8dccd46e30f99092.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 6]: /images/20220417/48d4ebc5d60248c7b646126cdeee5ba9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 7]: /images/20220417/f1ebedf5a2e945588cca9dff2a804bf6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 8]: /images/20220417/feb04af6ca5b4855a18e701486514115.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxMTAxMDIzNDk5MQ_size_16_color_FFFFFF_t_70 9]: /images/20220417/902411e8fc2e403095463cabe67ae499.png