Spark基础学习笔记32:Spark Streaming概述
Posted howard2005
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark基础学习笔记32:Spark Streaming概述相关的知识,希望对你有一定的参考价值。
文章目录
零、本讲学习目标
- 了解Spark Streaming的基本概念
- 了解Spark Streaming的使用场景
- 掌握Spark Streaming的工作原理
一、Spark Streaming概述
(一)什么是Spark Streaming
- Spark Streaming是Spark Core API(Spark RDD)的扩展,支持对实时数据流进行可伸缩、高吞吐量及容错处理。数据可以从Kafka、Flume、Kinesis或TCP Socket等多种来源获取,并且可以使用复杂的算法处理数据,这些算法由map()、reduce()、join()和window()等高级函数表示。处理后的数据可以推送到文件系统、数据库等存储系统。事实上,可以将Spark的机器学习和图形处理算法应用于数据流。
(二)Sparing Streaming的主要优点
- 使用Spark Streaming可以很容易地构建可伸缩的、容错的流应用程序。
1、易于使用
- Spark Streaming提供了很多高级操作算子,允许以编写批处理作业的方式编写流式作业。它支持Java、Scala和Python语言。
2、易于与Spark体系整合
- 通过在Spark Core上运行Spark Streaming,可以在Spark Streaming中使用与Spark RDD相同的代码进行批处理,构建强大的交互应用程序,而不仅仅是数据分析。
二、Spark Streaming工作原理
(一)Spark Streaming工作流程图
- Spark Streaming接收实时输入的数据流,并将数据流以时间片(
秒级
)为单位拆分成批次,然后将每个批次交给Spark引擎(Spark Core)进行处理,最终生成以批次组成的结果数据流。
(二)分段流 - DSteam(Discretized Stream)
1、分段流的概念
- Spark Streaming提供了一种高级抽象,称为DStream(Discretized Stream)。DStream表示一个连续不断的数据流,它可以从Kafka、Flume和Kinesis等数据源的输入数据流创建,也可以通过对其他DStream应用高级函数(例如map()、reduce()、join()和window())进行转换创建。
2、分段流的实质
- 在内部,对输入数据流拆分成的每个批次实际上是一个RDD,一个DStream由多个RDD组成,相当于一个RDD序列。
3、分段流中的RDD
- DStream中的每个RDD都包含来自特定时间间隔的数据
4、分段流的操作
- 应用于DStream上的任何操作实际上都是对底层RDD的操作。例如,对一个DStream应用
flatMap()
算子操作,实际上是对DStream中每个时间段的RDD都执行一次flatMap()
算子,生成对应时间段的新RDD,所有的新RDD组成了一个新DStream。对DStream中的RDD的转换是由Spark Core引擎实现的,Spark Streaming对Spark Core进行了封装,提供了非常方便的高层次API。
(三)输入DStream与Receiver
1、输入DStream与Receiver的关系
- 输入DStream表示从数据源接收的输入数据流,每个输入DStream(除了文件数据流之外)都与一个Receiver对象相关联,该对象接收来自数据源的数据并将其存储在Spark的内存中进行处理。
- 如果希望在Spark Streaming应用程序中并行接收多个数据流,那么可以创建多个输入DStream,同时将创建多个Receiver,接收多个数据流。但需要注意的是,一个Spark Streaming应用程序的Executor是一个长时间运行的任务,它会占用分配给SparkStreaming应用程序的一个CPU内核(占用Spark Streaming应用程序所在节点的一个CPU内核),因此Spark Streaming应用程序需要分配足够的内核(如果在本地运行,就是线程)来处理接收到的数据,并运行Receiver。
2、本地运行Spark Streaming应用程序
- 在本地运行Spark Streaming应用程序时,不要使用local或local[1]作为主URL。这两种方式都意味着只有一个线程将用于本地运行任务。如果正在使用基于Receiver的输入DStream(例如Socket、Kafka、Flume等),那么将使用单线程运行Receiver,导致没有多余的线程来处理接收到的数据(Spark Streaming至少需要两个线程:一个线程用于运行Receiver接收数据;另一个线程用于处理接收到的数据)。因此,在本地运行时,应该使用“local[n]”作为主URL,其中n大于Receiver的数量(若Spark Streaming应用程序只创建了一个DStream,则只有一个Receiver,n的最小值为2)。
3、集群上运行Spark Streaming应用程序
- 每个Spark应用程序都有各自独立的一个或多个Executor进程负责执行任务。将Spark Streaming应用程序发布到集群上运行时,每个Executor进程所分配的CPU内核数量必须大于Receiver的数量,因为1个Receiver独占1个CPU内核,还需要至少1个CPU内核进行数据的处理,这样才能保证至少两个线程同时进行(一个线程用于运行Receiver接收数据,另一个线程用于处理接收到的数据)。否则系统将接收数据,但无法进行处理。若Spark Streaming应用程序只创建了一个DStream,则只有一个Receiver,Executor所分配的CPU内核数量的最小值为2。
三、利用Spark Streaming完成词频统计
(一)提出任务
- 假设需要监听
TCP Socket
端口的数据,实时计算接收到的文本数据中的单词数
(二)完成任务
1、在master虚拟上安装nc
- nc是一个简单而有用的工具,透过使用TCP或UDP协议的网络连接去读写数据。它被设计成一个稳定的后门工具,能够直接由其它程序和脚本轻松驱动。同时,它也是一个功能强大的网络调试和探测工具,能够建立你需要的几乎所有类型的网络连接。
- 执行命令:
yum -y install nc
2、创建Maven项目
- 创建Maven项目 -
SparkStreamingWordCount
- 将
java
目录改成scala
目录
3、添加依赖与构建插件
- 在
pom.xml
文件里添加依赖与构建插件
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>net.hw.stream</groupId>
<artifactId>SparkStreamingWordCount</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.12</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.1.1</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.3.2</version>
<executions>
<execution>
<id>scala-compile-first</id>
<phase>process-resources</phase>
<goals>
<goal>add-source</goal>
<goal>compile</goal>
</goals>
</execution>
<execution>
<id>scala-test-compile</id>
<phase>process-test-resources</phase>
<goals>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
4、创建日志属性文件
- 在
resources
目录里创建log4j.properties
文件
log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
5、创建流式词频统计单例对象
- 创建
net.hw.stream
包
- 在
net.hw.stream
包里创建SparkStreamingWordCount
单例对象
package net.hw.stream
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds, StreamingContext
/**
* 功能:利用Spark Streaming实现词频统计
* 作者:华卫
* 日期:2022年05月17日
*/
object SparkStreamingWordCount
def main(args: Array[String]): Unit =
// 创建Spark配置对象
val conf = new SparkConf()
.setMaster("local[2]")
.setAppName("NetworkWordCount")
// 按照时间间隔为3秒钟切分数据流
val ssc = new StreamingContext(conf, Seconds(3))
// 创建行分段流(接收master节点9999端口的文本流数据)
val lines = ssc.socketTextStream("master", 9999)
// 生成单词分段流
val words = lines.flatMap(_.split(" "))
// 计算每一批次中的每个单词数量
val pairs = words.map((_, 1))
// 进行词频统计
val wc = pairs.reduceByKey(_ + _)
// 输出分段流中每个RDD的词频统计结果
wc.print()
// 开始计算
ssc.start()
// 等待计算结束
ssc.awaitTermination()
6、在master虚拟机上启动nc
- 执行命令:
nc -lk 9999
- 等待用户输入一行又一行的数据
7、启动程序,查看结果
- 启动流式词频统计单例对象,然后在master虚拟机上输入一行又一行的数据
- 因为强行停止程序,所以退出码是
-1
- 可以看到,每隔3秒,对分段流中的RDD进行词频统计……
以上是关于Spark基础学习笔记32:Spark Streaming概述的主要内容,如果未能解决你的问题,请参考以下文章
Spark基础学习笔记33:Spark Streaming数据源