文章目录
-
- 1、概述
- 2、Basic Concepts
-
- 2.1 Maven 添加依赖
- 2.2 StreamingContext
- 2.3 Discretized Streams (DStreams)
- 2.4 Input DStreams、 Receiver
- 2.5 Transformations on DStreams
- 2.6 Output Operations on DStreams
- 3、运行过程
- 4、官网案例WordCount
1、概述
官网：http://spark.apache.org/docs/latest/streaming-programming-guide.html
Spark流是核心Spark API的扩展，支持对实时数据流进行可伸缩的、高吞吐量的、容错的流处理。数据可以从Kafka、Flume、Kinesis或TCP套接字等多种来源获取，并且可以使用复杂的算法处理数据，这些算法由map、reduce、join和window等高级函数表示。最后，处理后的数据可以推送到文件系统、数据库和活动仪表板。
在内部，它的工作原理如下。Spark streams接收实时输入的数据流，并将数据分成批次，然后由Spark引擎对这些数据进行处理，以批量生成最终的结果流。
Spark流提供了一个高级抽象，称为discretized stream或DStream，它表示连续的数据流。DStreams可以从Kafka、Flume和Kinesis等源的输入数据流创建，也可以通过对其他DStreams应用高级操作创建。在内部，DStream表示为RDDs序列。
所以SparkStreaming严格意义上来讲并不是实时的，是由很多个小批次组成。
2、Basic Concepts
2.1 Maven 添加依赖
如果是基于Maven的Project，pom.xml要添加spark-streaming依赖包，注意scala版本，我的是scala-2.11.8
<spark.version>2.4.0</spark.version>
...
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
2.2 StreamingContext
StreamingContext是SparkStreaming程序的入口，回顾下前面讲的：
RDD入口 => SparkContext
DataFrame/DataSet入口 => SparkSession
DStream入口 => StreamingContextRDD入口 => SparkContext
DataFrame/DataSet入口 => SparkSession
DStream入口 => StreamingContext
import org.apache.spark._
import org.apache.spark.streaming._
val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1)) //conf不讲，Seconds表示多少秒执行一次程序
//TODO...要执行的代码
ssc.start()
ssc.awaitTermination()
2.3 Discretized Streams (DStreams)
Discretized Stream or DStream是Spark流提供的基本抽象。它表示连续的数据流，无论是从源接收到的输入数据流，还是通过转换输入流生成的经过处理的数据流。在内部，DStream由一系列连续的RDDs表示，RDDs是Spark对不可变的分布式数据集的抽象。DStream中的每个RDD都包含来自特定时间间隔的数据，如下图所示。
应用于DStream上的任何操作都转换为底层RDDs上的操作。
这些底层的RDD转换由Spark引擎计算。DStream操作隐藏了这些细节中的大部分，并为开发人员提供了更高级的API，以方便开发人员使用。
2.4 Input DStreams、 Receiver
Input DStreams是表示从流源接收的输入数据流的数据流。在官网案例中(该案例往下看)，lines是一个Input DStreams，因为它表示从netcat服务器接收到的数据流。每个Input DStreams都与Receiver对象相关联，后者接收来自源的数据并将其存储在Spark内存中进行处理。
注意，如果希望在流应用程序中并行接收多个数据流，可以创建多个input DStreams 。这将创建多个receivers，同时接收多个数据流。但是请注意，Spark worker或者executor端是一个长时间运行的任务，因此它占用分配给Spark流应用程序的一个Core。因此，重要的是要记住，Spark流应用程序需要分配足够的内核来处理接收到的数据，并运行receiver。
这段话是说如果你是Local模式的，那么至少要设置成
local[2]
，因为executor要占用1core，receiver接受数据源也要占用1core。所以像
local
，
local[1]
是不行的。
2.5 Transformations on DStreams
和SparkCore的算子使用是一样的，这里不讲解。
参考博客：https://blog.csdn.net/greenplum_xiaofan/article/details/97975195
2.6 Output Operations on DStreams
重点是foreachRDD
3、运行过程
4、官网案例WordCount
先开启natcat
[[email protected] bin]$ nc -lk 8888
再开启SparkStreaming应用程序，从natcat接受数据，每隔10秒一个批次计算WordCount
package com.ruozedata.spark
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
object SocketWCApp {
def main(args: Array[String]): Unit = {
val sparkConf=new SparkConf().setMaster("local[2]").setAppName("SocketWCApp")
val ssc=new StreamingContext(sparkConf,Seconds(10))
//socketTextStream底层源码里面就有 Receiver接收器
val lines=ssc.socketTextStream("vm01",8888)
val result=lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
result.print()
ssc.start()
ssc.awaitTermination()
}
}
在什么都没数据接收过来的情况下，每隔10秒会运行一次
要停止程序，点击左边红色正方形
然后输入一些数据
[[email protected] bin]$ nc -lk 8888
hello hadoop
hello spark