Spark大数据技术基础与应用（Scala+Python版）课件 第5章 Spark数据流处理

Spark流数据处理第5章本章学习目标030201掌握两种流处理框架的基本概念和运行机理熟悉两种流处理框架的基本使用方法理解流处理中的容错机制、状态管理以及时间处理策略等高级特性045.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming的产生SparkStreaming是构建在Spark上的实时计算框架扩展了Spark处理大规模流式数据的能力SparkStreaming可结合批处理和交互查询，适合一些

需要对历史数据和实时数据进行结合分析的应用场景5.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming的工作原理SparkStreaming支持从多种数据源提取数据，如Kafka、Flume、Twitter、ZeroMQ、文本文件以及TCP套接字等；并且可以提供一些高级API来表达复杂的处理算法，如map、reduce、join和window等；SparkStreaming支持将处理完的数据推送到文件系统、数据库或者实时仪表盘中展示。5.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming的工作原理对于流数据，SparkStreaming接收实时输入的数据流后，将数据流按照时间片（秒级）为单位进行拆分为一个个小的批次数据，然后经Spark引擎以类似批处理的方式处理每个时间片数据。5.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming的工作原理DStream表⽰的是连续数据流，可以是源数据接收的输⼊流，也可以是通过转换输⼊流⽣成的已处理的数据流；在内部，DStream由⼀系列的RDD组成；DStream中的每个RDD都包含来⾃特定间隔的数据5.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming实时数据流处理的工作流程SparkStreaming的工作流程可以分为以下几个关键步骤：接收器（Receiver）：SparkStreaming的起点是接收器，它负责从各种数据源（如Kafka、Flume等）接收实时数据流。驱动器（Driver）：驱动器程序是SparkStreaming作业的入口点。在这里，定义了一个StreamingContext对象，它告诉Spark如何运行这个流计算任务。5.1SparkStreaming概述5.1.1SparkStreaming介绍SparkStreaming实时数据流处理的工作流程StreamingContext：在SparkStreaming中，所有的操作都是基于StreamingContext进行的。SparkContext：SparkContext是Spark作业的核心，负责整个应用的上下文管理。工作节点（WorkerNodes）：在Spark集群中，工作节点是执行实际计算任务的机器。执行器（Executors）：执行器是工作节点上运行的任务处理进程。长期运行任务（Long-RunningTasks）：SparkStreaming的设计目标是支持长期运行的任务。5.1SparkStreaming概述5.1.2第一个SparkStreaming程序SparkStreaming实时数据流处理的一般步骤引入必要的库

在项目中引入SparkStreaming相关的库。

如果是Python，则导入pyspark.streaming模块。例如：Scala：Python：5.1SparkStreaming概述5.1.2第一个SparkStreaming程序SparkStreaming实时数据流处理的一般步骤创建StreamingContext

创建StreamingContext对象，它是SparkStreaming的入口点。指定一个批处理间隔，它决定了多久处理一次数据流。其中第2个参数是批处理间隔。例如：Scala：Python5.1SparkStreaming概述5.1.2第一个SparkStreaming程序SparkStreaming实时数据流处理的一般步骤创建DStream：使用StreamingContext从数据源接收数据，并创建DStream。例如：Scala：Python5.1SparkStreaming概述5.1.2第一个SparkStreaming程序SparkStreaming实时数据流处理的一般步骤转换与操作：启动流处理：使用start()方法启动流处理。等待处理完成：使用awaitTermination()或awaitTerminationOrTimeout()方法等待流处理完成。停止StreamingContext：当不再需要流处理时，使用stop()方法停止StreamingContext。监控与调优：使用SparkUI或第三方监控工具监控流处理的性能。部署与扩展：根据需要部署SparkStreaming应用到集群上。5.2DStream基础操作5.2.1DStream的创建HDFS文件数据源文件数据流能够读取所有HDFSAPI兼容的文件系统文件，通过fileStream方法进行读取，SparkStreaming将会监控数据所在目录并不断处理移动进来的文件。文件数据流目前不支持嵌套目录。需要注意的是：1）文件需要有相同的数据格式；

2）文件进入数据所在目录的方式需要通过移动或

者重命名来实现；3）一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据；。5.2DStream基础操作5.2.1DStream的创建RDD队列创建DStream在ApacheSparkStreaming框架中，可以利用RDD队列来创建DStream，以模拟实时数据流或用于测试目的。每当一个新的RDD被添加到queueOfRDDs队列中时，SparkStreaming会将其视为一个新的数据批次，并按设定的批处理时间间隔执行相应的计算逻辑。5.2DStream基础操作5.2.2DStream的转换DStream无状态转换操作DStream进行各种转换操作，大致上可以分为两种：有状态转换和无状态转换。无状态转换操作是指那些不需要维护状态信息的转换操作，每个批次之间相互独立，不依赖于之前的数据处理结果。无状态转换操作非常适合于一些独立的数据转换和过滤，比如映射、过滤、扁平化等操作。5.2DStream基础操作5.2.2DStream的转换DStream有状态转换操作DStream有状态转换操作是指在SparkStreaming中，允许对DStream的数据进行跨时间批次的状态管理操作。这些操作不同于无状态转换，无状态转换只关注当前批次的数据，不会对之前批次的数据保持记忆。有状态转换操作在处理连续的流数据时，可以维护每个键（key）的状态，从而在多个批次之间保存和更新这些状态。有状态转换操作的主要目的是在处理数据流时跟踪和存储状态信息，这对于一些需要累积历史数据或跟踪历史状态的场景来说非常有用。5.2DStream基础操作5.2.2DStream的转换DStream的窗口操作在SparkStreaming中，为DStream提供窗口操作，即在DStream流上，将一个可配置的长度设置为窗口，以一个可配置的速率向前移动窗口。根据窗口操作，对窗口内的数据进行计算，每次落在窗口内的RDD数据会被聚合起来计算，生成的RDD会作为WindowDStream的一个RDD。5.2DStream基础操作5.2.2DStream的转换DStream的窗口操作DStream常用的窗口转换操作window(windowLength,slideInterval)：基于源DStream产生的窗口化的批数据计算一个

新的DStreamcountByWindow(windowLength,slideInterval)：返回流中元素的一个滑动窗口数reduceByWindow(func,windowLength,slideInterval)：返回一个单元素流reduceByKeyAndWindow(func,windowLength,slideInterval,[numTasks])：应用到一个(K,V)对组成的DStream上，返回一个由(K,V)对组成的新的DStream。reduceByKeyAndWindow(func,invFunc,windowLength,slideInterval,[numTasks])(6)countByValueAndWindow(windowLength,slideInterval,[numTasks])5.2DStream基础操作5.2.3DStream的输出DStream的输出是SparkStreaming处理数据流的最终步骤，它决定了如何将经过转换和计算后的数据流发送到外部系统或进行进一步的处理。在SparkStreaming中，DStream的输出通常涉及将处理后的数据持久化到外部存储系统、发送给其他处理系统或进行实时分析和可视化。输出操作通常会在DStream转换操作的最后阶段执行，例如经过map、filter、reduceByKey等转换之后。5.2DStream基础操作5.2.3DStream的输出DStream的输出是SparkStreaming处理数据流的最终步骤，它决定了如何将经过转换和计算后的数据流发送到外部系统或进行进一步的处理。在SparkStreaming中，DStream的输出通常涉及将处理后的数据持久化到外部存储系统、发送给其他处理系统或进行实时分析和可视化。输出操作通常会在DStream转换操作的最后阶段执行，例如经过map、filter、reduceByKey等转换之后。5.3StructuredStreaming结构化流5.3.1StructuredStreaming概述StructuredStreaming的核心在于将数据流视作一个持续更新的表（无界表UnboundedTable），新数据不断被追加至该表中。Spark结构化流将数据流抽象为动态增长的表，随着新数据的实时到达，该表持续扩展。编程模型5.3StructuredStreaming结构化流5.3.1StructuredStreaming概述结构化流有以下几种输出模式：完成模式（CompleteMode）：整个更新的结果表将写入外部存储器。追加模式（AppendMode）：只有自上一个触发器以来追加到结果表中的新行才会写入外部存储器。更新模式（UpdateMode）：只有自上次触发器以来在结果表中更新的行才会写入外部存储器（从spark2.1.1开始可用）。输出模式5.3StructuredStreaming结构化流5.3.1StructuredStreaming概述完成在StructuredStreaming中，事件时间（数据实际发生时间）对确保数据处理准确性至关重要，尤其是在跨时间窗口处理时。StructuredStreaming通过Watermark机制支持事件时间处理，Watermark标识了最早可能到达的延迟数据的时间戳。晚于Watermark的数据视为延迟数据，可能在后续窗口处理，这提升了系统处理延迟数据的鲁棒性和准确性。处理事件时间（Event-time）和延迟数据（LateData）5.3StructuredStreaming结构化流5.3.1StructuredStreaming概述Spark结构化流以其卓越的容错机制确保了数据流处理的可靠性与连续性。其从设计之初便融入了容错数据源的概念，即使遭遇节点故障或任务中断，也能无缝衔接，从上次读取点继续处理，避免数据丢失。状态管理方面，Spark结构化流利用检查点机制，将处理状态定期记录至可靠的存储，确保在任务重启或失败时，能够迅速恢复状态，保持数据处理的一致性。此外，Spark的任务调度与资源管理框架（如YARN、Mesos等）也提供了强大的容错能力，支持节点故障恢复与任务重试，保障资源的高效利用与系统的稳定运行。容错机制和语义5.3StructuredStreaming结构化流5.3.2StructuredStreaming操作StructuredStreaming内置的输入源Kafka源：要求ApacheKafka的版本是0.10或更高版本。文件源：文件位于本地文件系统、HDFS上。Socket源：这仅用于测试目的。Rate源：这仅用于测试和基准测试。5.3StructuredStreaming结构化流5.3.2StructuredStreaming操作StructuredStreaming内置的输出Sink数据接收器（DataSink）是用来存储流应用程序的输出的。不同的Sink可以支持不同的输出模式，并且具有不同的容错能力。在StructuredStreaming中，输出Sink是整个流处理流水线的终点，它是将经过实时计算后的结果数据写出到外部存储系统或服务的关键组件。Sink可以是各种类型的数据存储，如HDFS、数据库、消息队列（如Kafka）、甚至是实时Dashboard等。5.3StructuredStreaming结构化流5.3.2StructuredStreaming操作StructuredStreaming内置的输出Sink数据接收器（DataSink）是用来存储流应用程序的输出的。不同的Sink可以支持不同的输出模式，并且具有不同的容错能力。在StructuredStreaming中，输出Sink是整个流处理流水线的终点，它是将经过实时计算后的结果数据写出到外部存储系统或服务的关键组件。Sink可以是各种类型的数据存储，如HDFS、数据库、消息队列（如Kafka）、甚至是实时Dashboard等。目前StructuredStreaming内置FileSink、ConsoleSink、ForeachSink（ForeachBatchSink）、MemorySink及KafkaSink，其中测试最为方便的是ConsoleSink。5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性事件时间和处理时间在流处理中，涉及到两个时间概念：事件时间和处理时间(ProcessingTime)。事件时间，是事件在其生产设备上发生的时间。事件时间处理可用于无序事件。处理时间，是机器执行数据流处理的系统时间，是执行基于时间的操作的每个操作符的本地时间。原则：当应用程序遇到一些问题时，需要从前一个检查点或保存点重放它时，是否希望结果完全一样。如果想要结果完全相同，只能使用事件时间；如果接受不同的结果，可以使用处理时间。事件时间窗口5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性事件时间窗口概念在StructuredStreaming中，事件时间是指事件实际发生的物理时间，而非数据被系统处理的时间。基于事件时间的窗口操作允许按照事件时间将数据流分割成一个个连续的、不重叠或重叠的时间窗口，并在每个窗口上执行聚合操作。事件时间窗口5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性事件时间窗口类型固定窗口（TumblingWindow）：是一种特殊的窗口划分方式，它将数据流分成一系列不重叠的、固定长度的时间区间。每个窗口内的数据都是相互独立的，即一个事件只属于一个窗口。滑动窗口（SlidingWindow）：是基于时间长度划分的窗口，但它与固定窗口的不同之处在于窗口之间存在重叠部分。滑动窗口会连续不断地移动，并在每次移动后再次对新的时间区间内的数据进行计算。会话窗口（SessionWindow）：不同于固定窗口和滑动窗口，会话窗口不是基于固定时间间隔划分的，而是依据事件之间的活跃程度或者说会话活动间隙来动态定义窗口的边界。在会话窗口中，窗口的开始和结束时间由事件到达的时间间隔决定。事件时间窗口5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性水印的概念在实时数据流场景下，事件的发生时间并不总是与其到达处理系统的顺序一致，即可能存在乱序事件。水印是对事件时间的一种滞后标记，代表了数据流处理系统可以接受的最大延迟界限。水印是一个动态更新的阈值，它反映了系统当前处理到的事件时间戳集合中，可以容忍的最大延迟时间点。水印机制在流处理框架（如ApacheFlink、SparkStreaming等）中扮演着关键角色，它有助于在保证数据处理时效性的同时，尽可能保证数据处理结果的完整性。水印5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性水印的作用窗口触发计算:当水印时间戳向前推进时，那些事件时间早于或等于当前水印的所有窗口都会触发计算，从而输出窗口内的汇总结果。乱序事件处理:水印机制允许一定范围内的乱序事件参与计算，超出水印阈值的迟到事件可能无法影响已经完成计算的窗口，但它们仍有可能参与到尚未关闭的窗口或者后续窗口的计算中。数据一致性保障:尽管不能完全避免乱序带来的影响，但通过合理设置水印，可以在乱序容忍度和数据一致性之间取得平衡，确保在面对乱序事件时系统能给出尽可能准确且及时的结果。水印5.3StructuredStreaming结构化流5.3.3StructuredStreaming高级特性容错数据源：Spark结构化流可以从容错的数据源读取数据。当某个任务失败时，Spark会自动从上次读取的位置继续读取数据，并确保不丢失任何数据。容错数据处理：Spark结构化流使用基于事件时间的处理方式，可以处理延迟数据和乱序数据。容错状态管理：在具有状态的计算中，Spark结构化流提供了容错的状态管理机制。任务调度和资源管理：Spark结构化流使用Spark的任务调度和资源管理机制，例如SparkStandalone、ApacheMesos或ApacheHadoopYARN等。容错机制和语义5.4SparkStreaming编程实例5.4.1项目介绍随着大数据的发展与应用，互联网设备日益扩增。在集群模式下人工难以维护，为解决此类问题，智能物联网系统成为目前趋势。通过实时数据监控，减轻运维人员工作量，节约公司成本。因此我们通过Spark技术，模拟对物联网设备进行监控，实时统计分析设备数据信息，从而完成智能运维工作。项目背景5.4SparkStreaming编程实例5.4.1项目介绍模拟数据生成

为较为全面展示智能物联网系统开发，在测试中常常会通过模拟数据生成手段来进行测试开发。消息队列Kafka

项目中将生成数据通过消息队列Kafka进行消费分析，其目的防止上下游数据到达顶峰导致数据出错。数据处理分析Spark

本项目通过StructuredStreaming实现对流数据实时监控分析，通过指定指标分析设备运营情况。编程语言

基于Scala进行Spark程序编写，因其面向函数式编程，在代码效率上优越于java程序。因此选用Scala作为程序开发中的编程语言。项目架构5.4SparkStreaming编程实例5.4.1项目介绍

模拟一个智能物联网系统的数据统计分析，产生设备数据发送到Kafka，结构化流StructuredStreaming实时消费统计。对物联网设备状态信号数据实时统计分析。统计指标：（1）信号强度大于30的设备。（2）各种设备类型的数量。（3）各种设备类型的平均信号强度。项目需求5.4SparkStreaming编程实例5.4.2知识点介绍设备监控日志数据如下所示，我们须通过模拟数据生成并发送Kafka。数据以json格式传输，包含设备标识ID、设备类型、设备信号、发送数据时间4个字段。数据解析5.4SparkStreaming编程实例5.4.2知识点介绍（1）Kafka创建生产者消费者主题，用于数据消费。（2）Scala编程模拟数据生成，并将数据发送到Kafka主题。（3）通过Spark从Kafka主题消费数据，对数据进行实时分析。（4）通过控制台监控分析结果数据。项目流程5.4SparkStreaming编程实例5.4.2知识点介绍模拟数据生成：Random函数方便随机数的产生，可以通过该函数快捷得到我们想要的数据。定义random对象如下所示：Kafka读取或消费数据：Kafka提供生产者/消费者API有效帮助我们实现数据的读取或消费，只需要设置Kafka集群地址、主题、kv序列化等即可实现数据发送。项目技术点5.4SparkStreaming编程实例5.4.2知识点介绍

发送数据到Kafka：

从Kafka消费数据：项目技术点5.4SparkStreaming编程实例5.4.3项目实现主题准备5.4SparkStreaming编程实例5.4.3项目实现（1）项目名：structuredstreaming。（2）包结构：com.hongya.structuredstreaming。（3）实现类：模拟数据生成实现：MockIotDatas。日志数据分析处理实现：Streaming_Iot。项目创建5.4SparkStreaming编程实例5.4.3项目实现模拟数据生成并发送Kafka：（1）定义样例类DeviceData存储基站通话日志数据。（2）在MockIotDatas中定义main方法，在main方法中实现业务逻辑。（3）通过创建Properties文件配置类对象props设置发送Kafka主题参数。通过put方法设置Kafka生产者集群地址。通过put方法设置ack机制为1。通过put参数设置生产者重新发送消息次数为3。通过put参数设置生产者key，value序列化方式为String类型。创建Kafka生产者对象producer传入文件参数props。实现步骤5.4SparkStreaming编程实例5.4.3项目实现（4）模拟物联网数据生成发送Kafka：

定义random随机数对象。

定义deviceTypes数组变量存放通话状态信息，存储元素值为Array("mysql","redis","kafka","route","redis","flume","mysql","kafka","mysql")。实现步骤5.4SparkStreaming编程实例5.4.3项目实现通过while无限循环模拟数据生成。

定义index变量获取数组deviceTypes数组长度区间随机值。

定义deviceId变量获取设备标识符ID为s"device_${(index+1)*10+random.nextInt(index+1)}"。

定义变量deviceType获取设备类型数据为devicTypes数组中下标为index的随机值。

定义变量deviceSignal获取设备信号值为10+random.nextInt(90)。

通过样例类DeviceData将数据进行封装，发送时间为当前系统时间。

将封装后数据返回给deviceData变量。

实现步骤5.4SparkStreaming编程实例5.4.3项目实现通过while无限循环模拟数据生成。

将deviceData转为json数据类型deviceJson。

打印deviceJson。

线程休眠，时间为1000+random.nextInt(500)。

创建ProducerRecord对象record发送生产者主题以及数据参数。通过producer对象调用send方法发送主题和数据信息record。实现步骤5.4SparkStreaming编程实例5.4.3项目实现关闭连接从Kafka消费数据并按需求指标统计分析：（1）在Streaming_Iot中定义main方法，在main方法中实现业务逻辑。（2）创建环境：

创建应用入口SparkSession对象spark，设置应用名称、本地运行模式以及Shuffle分区数目为3。

通过spark对象获取sparkContext对象sc，设置日志级别WARN。

导入隐式转换。

导入内置函数包。实现步骤5.4SparkStreaming编程实例5.4.3项目实现（3）从Kafka读取数据：

通过spark.readStream进行读取。

通过format方法设置从kafka读取。

通过option方法设置kafka集群主机名qngjiao和对应端口号9092。

通过option方法设置生产者主题iotTopic。

通过option方法设置每批次消费数据最大值为100000。

通过load方法加载数据，将数据返回给iotStreamDF。实现步骤5.4S