Spark Streaming连接TCP Socket

时间:2021-09-19 00:46:58

1.Spark Streaming是什么

Spark Streaming是在Spark上建立的可扩展的高吞吐量实时处理流数据的框架,数据可以是来自多种不同的源,例如kafka,Flume,Twitter,ZeroMQ或者TCP Socket等。在这个框架下,支持对流数据的各种运算,比如map,reduce,join等。处理过后的数据可以存储到文件系统或数据库。

利用Spark Streaming,你可以使用与批量加载数据相同的API来创建数据管道,并通过数据管道处理流式数据。此外,Spark Steaming的“micro-batching”方式提供相当好的弹性来应对某些原因造成的任务失败。

2. Spark Streaming的基本原理

Spark Streaming对数据的处理方式主要采用的方法是对Stream数据进行时间切片,分成小的数据片段,通过类似批处理的方式处理数据片段。

Spark Streaming把实时输入数据流以时间片Δt (如1秒)为单位切分成块。Spark Streaming会把每块数据作为一个RDD,并使用RDD操作处理每一小块数据。

Spark Streaming将流式计算分解成一系列短小的批处理作业。Spark Streaming的输入数据分成一段一段的数据(DStreaming),每一段数据都转换成Spark中的RDD,然后将Spark Streaming中对DStream的操作变为针对Spark中对RDD的操作,将RDD经过操作变成中间结果保存在内存中。

3. DStream

上面提到了DStreaming,那么DStreaming到底是什么呢:

DStreaming相当于在Streaming的框架下对RDD进行封装,表示的是我们处理的一个实时数据流。类似于RDD,DStream提供了转换操作,窗口转换操作和输出操作三种操作方法。

4.Spark Streaming的优势

Spark Streaming是一种构建在Spark上的实时计算框架,它扩展了Spark处理大规模流式数据的能力。

实时性:能运行在100+的结点上,并达到秒级延迟。Spark Streaming将流式计算分解成多个Spark Job,对于每一段数据的处理都会经过Spark的任务集的调度过程。其最小的Batch Size的选取在0.5~2秒钟之间(Storm目前最小的延迟是100ms左右),所以Spark Streaming能够满足除对实时性要求非常高的所有流式准实时计算场景。

高效和容错的特性:对于流式计算来说,容错性至关重要。在spark中每一个RDD都是一个不可变的分布式可重算的数据集,其记录着确定性的操作,只要输入数据是可容错的,那么任意一个RDD的分区出错或不可用,都是可以利用原始输入数据通过转换操作而重新算出的。而spark Streaming使用基于内存的Spark作为执行引擎, 其容错性自然很好。

吞吐量:Spark Streaming能集成Spark的批处理和交互查询,其吞吐量比Storm至少高2~5倍。并且它为实现复杂的算法提供了和批处理类似的简单接口。

 

接下来用Spark  Streaming连接TCP Socket来说明如何使用Spark  Streaming:

1 创建StreamingContext对象

首先使用StreamingContext模块,这个模块的作用是提供所有的流数据处理的功能:

1 from pyspark import SparkContext
2 from pyspark.streaming import StreamingContext
3 
4 sc = SparkContext("local[2]", "streamwordcount")
5 # 创建本地的SparkContext对象,包含2个执行线程
6 
7 ssc = StreamingContext(sc, 2)
8 # 创建本地的StreamingContext对象,处理的时间片间隔时间,设置为2s

2 创建DStream对象

我们需要连接一个打开的 TCP 服务端口,从而获取流数据,这里使用的源是TCP Socket,所以使用socketTextStream()函数:

lines = ssc.socketTextStream("localhost", 8888)
# 创建DStream,指明数据源为socket:来自localhost本机的8888端口

3 对DStream进行操作

我们开始对lines进行处理,首先对当前2秒内获取的数据进行分割并执行标准的MapReduce流程计算。

words = lines.flatMap(lambda line: line.split(" "))
# 使用flatMap和Split对2秒内收到的字符串进行分割

得到的words是一系列的单词,再执行下面的操作:

pairs = words.map(lambda word: (word, 1))
# map操作将独立的单词映射到(word,1)元组

wordCounts = pairs.reduceByKey(lambda x, y: x + y)
# reduceByKey操作对pairs执行reduce操作获得(单词,词频)元组

5 输出数据

将处理后的数据输出到一个文件中:

outputFile = "/home/feige/streaming/ss"
# 输出文件夹的前缀,Spark Streaming会自动使用当前时间戳来生成不同的文件夹名称

wordCounts.saveAsTextFiles(outputFile)
# 将结果输出

6 启动应用

要使程序在Spark Streaming上运行起来,需要执行Spark Streaming启动的流程,调用start()函数启动,awaitTermination()函数等待处理结束的信号。

ssc.start() 
# 启动Spark Streaming应用
ssc.awaitTermination()

打开终端执行:

nc -lk 8888

nc的-l参数表示创建一个监听端口,等待新的连接。-k参数表示当前连接结束后仍然保持监听,必须与-l参数同时使用。

执行完上面的命令后不关闭终端,我们将在这个终端中输入一些处理的数据:

Spark Streaming连接TCP Socket

打开一个新的终端来执行我们的Spark Streaming应用:

Spark Streaming连接TCP Socket

 这里是spark streaming执行的过程

现在我们来看看程序执行的效果,程序每隔2秒扫描一次监控窗口输入的内容,我们查看一下:

Spark Streaming连接TCP Socket

结束语:

最近压力比较大,琐事诸多,相信这段时间过后一切都会好起来的,加油!!!