Spark Streaming简介

电玩女神 2023-03-01 06:07 156阅读 0赞

参考：[Spark Streaming介绍][Spark Streaming]

[SparkStreaming 介绍][SparkStreaming]

[SparkStreaming官方文档][SparkStreaming 1]

## Spark Streaming 简介 ##

Spark Streaming是一个基于Spark Core之上的实时计算框架，可以从很多数据源消费数据并对数据进行实时的处理，具有高吞吐量和容错能力强等特点。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jidmp4MTMxNA_size_16_color_FFFFFF_t_70][]

## Spark Streaming 特点： ##

>  *  **高吞吐量**：Streaming 在 Spark 的基础上集成了流式处理，可以以类似 Spark 批处理的方式写流式作业，"接收+处理+输出"大量数据。一个吞吐，可以说是，一个服务器接受客户端的请求==》然后处理完（可能是 CPU 计算、可能是文件处理、数据库处理、网络任务等）==》最后返回结果。
>  *  **容错能力强**：可以恢复丢失的作业及操作状态
>  *  **支持多数据源输入**：hdfs、flume、kafka、twitter、zeromq 和自定义数据源读取数据，做交互应用、数据分析
>  *  **结果存储**：能保存在很多地方，如 HDFS，数据库等
>  *  能和 MLlib（机器学习）以及 Graphx 完美融合

## 构建一个 Spark Streaming 应用程序的步骤： ##

> 1.  构建 Streaming Context 对象时，注意**时间窗口参数** Second(1)，需要积攒的时间长度，如1s。该参数设置依赖于需求和集群处理能力。
> 2.  创建 InputDStream
> 3.  操作 DStream
> 4.  启动 Spark Streaming，当 ssc.start() 启动后，程序才真正进行所有预期的操作

离散数据流 (DStream)：在其内部，DStream 是通过一组时间序列上连续的 RDD 来表示的，每一个 RDD 都包含了特定时间间隔内的数据流

**相关算子**

>  *  map(func)
>  *  flatMap(func)
>  *  reduceByKey(func,numTasks)：numTasks 并行的提交任务个数，默认情况下，本地环境下是 2，集群环境下是 8
>  *  foreachRDD(func)：基本的输出操作

## Spark Streaming 接收 Kafka 数据： ##

*  先把数据接收过来，转换为spark streaming中的数据结构Dstream，有两种方式：利用Receiver接收数据 OR 直接从kafka读取数据

> **方式一 基于Receiver的方式：**
> 
>  *  对于所有的接收器，从kafka接收来的数据会存储在 Spark 的 exector 中，之后 Streaming 提交的 job 会处理这些数据
>  *  **注意：**Receiver的方式中，Spark 的partition 和 Kafka 的 partition 并不相关，加大每个 topic 的 partition 仅仅增加线程来处理单一 Receiver 消费的 topic，并没有增加 Spark 在处理数据上的并行度
>  *  对于不同的 Group 和 topic 可以使用多个Receiver创建不同的DStream来并行接收数据，之后可以用union来统一成一个DStream
>  *  如果我们启用了 Write Ahead Logs 复制到文件系统如 HDFS，那么 storage level 需要设置成 StorageLevel.MEMORY\_AND\_DISK\_SER，也就是`KafkaUtils.createStream(..., StorageLevel.MEMORY_AND_DISK_SER)`
> 
> **方式二 直接读取方式：**
> 
>  *  Spark1.3 之后，引入了 Direct 方式，它会周期性的获取 Kafka 中每个 topic 的每个 partition 中的最新 offsets,之后根据设定的 maxRatePerPartition 来处理每个 batch
>  *  **简化的并行：**在\*\*Direct方式中，Kafka中的partition与RDD中的partition是一一对应的并行读取Kafka数据，这种映射关系也更利于理解和优化
>  *  **高效：**在Receiver的方式中，为了达到0数据丢失需要将数据存入Write Ahead Log中，这样在Kafka和日志中就保存了两份数据，浪费！而第二种方式不存在这个问题，只要我们Kafka的数据保留时间足够长，我们都能够从Kafka进行数据恢复
>  *  **精确一次消费：**在Receiver的方式中，使用的是Kafka的高阶API接口从Zookeeper中获取offset值，这也是传统的从Kafka中读取数据的方式，但由于Spark Streaming消费的数据和Zookeeper中记录的offset不同步，这种方式偶尔会造成数据重复消费。而第二种方式，直接使用了简单的低阶Kafka API，Offsets则利用Spark Streaming的checkpoints进行记录，消除了这种不一致性

## Spark streaming+Kafka调优方法 ##

参考：[https://www.cnblogs.com/xltur...][https_www.cnblogs.com_xltur...]

目的是，Spark Streaming能够实时的拉取Kafka当中的数据，并且能够保持稳定

> 1.  合理的批处理时间（batchDuration）: 过小会导致数据处理不完，导致阻塞。**一般对于batchDuration的设置不会小于500ms**。
> 2.  合理的Kafka拉取量，配置参数为spark.streaming.kafka.maxRatePerPartition：默认是没有上线的，即kafka当中有多少数据它就会直接全部拉出。而根据生产者写入Kafka的速率以及消费者本身处理数据的速度，同时这个参数需要结合上面的batchDuration，使得每个partition拉取在每个batchDuration期间拉取的数据**能够顺利的处理完毕，做到尽可能高的吞吐量**，而这个参数的调整可以参考可视化监控界面中的Input Rate和Processing Time
> 3.  缓存反复使用的 DStream（RDD）：Spark 中的 RDD 和SparkStreaming 中的 Dstream，**如果被反复的使用，最好利用cache()**，将该数据流缓存起来，防止过度的调度资源造成的网络开销
> 4.  设置合理的GC：即java 的垃圾回收机制，让我们不用关注内存的分配回收，更加专注业务逻辑。java 的 GC 分为初生代、年轻代、老年代、永久代，每次GC都是需要耗费时间的，特别是老年代的GC回收。**通常使用中建议：** `--conf "spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC"`
> 5.  设置合理的 CPU 资源数：**CPU 的 core 数量，每个 exectutor** 可以占用多个 core，但是 executor 并不总能充分利用多核的能力。通过观察 CPU 的资源使用情况，可以适量增加 executor 减少 core 数量，注意 executor 增加，每个 exector 的内存就越小哦（executor太多的话，出现过多数据时就会出现 spill over 甚至 out of memory）！
> 6.  设置合理的parallelism：在SparkStreaming+kafka的使用中，我们采用了Direct连接方式，而Spark中的partition和Kafka中的Partition是一一对应的，我们一**般默认设置为Kafka中Partition的数量**
> 7.  **使用Kryo优化序列化性能**：官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右
> 8.  **使用高性能的算子：**
> 
>  *  使用reduceByKey/aggregateByKey替代groupByKey
>  *  使用mapPartitions替代普通map
>  *  使用foreachPartitions替代foreach
>  *  使用filter之后进行coalesce操作
>  *  使用repartitionAndSortWithinPartitions替代repartition与sort类操作

// 创建SparkConf对象。
    val conf = new SparkConf().setMaster(...).setAppName(...)
    // 设置序列化器为KryoSerializer。
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    // 注册要序列化的自定义类型。
    conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

[Spark Streaming]: https://segmentfault.com/a/1190000020667887
[SparkStreaming]: https://blog.csdn.net/bbvjx1314/article/details/105515871
[SparkStreaming 1]: http://spark.apache.org/docs/latest/streaming-programming-guide.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Jidmp4MTMxNA_size_16_color_FFFFFF_t_70]: /images/20230209/5140fd9aaa0f4faaa744d108b47541ba.png
[https_www.cnblogs.com_xltur...]: https://www.cnblogs.com/xlturing/p/6246538.html