你好,游客 登录
背景:
阅读新闻

Spark 2.3.0 Spark Streaming Programming Guide 学习笔记

[日期:2018-09-01] 来源:csdn  作者:hjbbjh0521 [字体: ]

一 概述

spark是近实时的流处理框架,支持的数据源有kafka、flume、kinesis、tcp sockets、文件系统等。流式读取数据后,可以用类似map、reduce、join和window等高层函数进行处理。最终,处理后的数据可以写入文件系统、数据库、实时仪表盘等。这里其实已经把流式数据抽象成了一个个小批次的分布式数据集,因此,你也可以在这些数据之上进行机器学习以及图计算。

内部实现如下图(把流式数据分成一个个小的批次数据):

spark streaming提供一个DStream的抽象概念,代表一个源源不断的数据流。DStream可以从上面提到的数据源得到也可以从其他的DStream得到,DSteam其实就是代表很多RDD的集合。

二 一个小例子

监听tcp socket,接收流式数据,做单词统计。

首先要导入一些包,StreamingContext是spark streaming主要的入口类,下面我们创建了一个本地的StreamingContext(两个执行线程),每间隔1秒一个批次。

  1. import org.apache.spark._
  2. import org.apache.spark.streaming._
  3. import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
  4.  
  5. // Create a local StreamingContext with two working thread and batch interval of 1 second.
  6. // The master requires 2 cores to prevent a starvation scenario.
  7.  
  8. val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
  9. val ssc = new StreamingContext(conf, Seconds(1))

然后可以指定监听tcp端口(地址和端口)创建DStream,例如(本地9999端口):

  1. // Create a DStream that will connect to hostname:port, like localhost:9999
  2. val lines = ssc.socketTextStream("localhost", 9999)

lines其实就是从这个端口接收的文本数据,一行一个记录(多个RDD),然后把行记录分隔成一个单词一个记录:

  1. // Split each line into words
  2. val words = lines.flatMap(_.split(" "))

然后就是对这些RDD集进行处理,得到单词计数:

  1. import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
  2. // Count each word in each batch
  3. val pairs = words.map(word => (word, 1))
  4. val wordCounts = pairs.reduceByKey(_ + _)
  5.  
  6. // Print the first ten elements of each RDD generated in this DStream to the console
  7. wordCounts.print()

以上是一个批次的单词计数,也就是上面我们创建的时候指定的1秒。

但仅仅上面那些代码,运行后不会进行计算,要通过下面代码启动计数:

  1. ssc.start() // Start the computation
  2. ssc.awaitTermination() // Wait for the computation to terminate

三 基本概念

3.1 添加依赖

maven依赖:
  1. <dependency>
  2. <groupId>org.apache.spark</groupId>
  3. <artifactId>spark-streaming_2.11</artifactId>
  4. <version>2.3.0</version>
  5. </dependency>

仅仅添加上面的依赖,是不支持类似kafka、flume、kinesis等数据源的,需要添加类似spark-streaming-xyz_2.11这样的依赖到自己的工程中:

3.2 初始化StreamingContext

  1. import org.apache.spark._
  2. import org.apache.spark.streaming._
  3.  
  4. val conf = new SparkConf().setAppName(appName).setMaster(master)
  5. val ssc = new StreamingContext(conf, Seconds(1))

可以上面方式进行初始化,appname会显示到UI上,master可以设置为spark、mesos或yarn集群方式的url,或者用local[*]字符串来指定为本地模式。实际上不会硬编码到程序里,通常会通过spark-submit命令行进行指定。如果你想要使用SparkContext,那么可以通过ssc.sparkContext来访问。

流数据的批次的时间间隔,需要根据你应用的实际需求以及集群资源来确定,详细可参考性能调节章节。

同样StreamingContext也可以通过一个已经存在的SparkContext来创建:

  1. import org.apache.spark.streaming._
  2.  
  3. val sc = ... // existing SparkContext
  4. val ssc = new StreamingContext(sc, Seconds(1))

在context创建之后,你还要做如下步骤:

1、定义创建DStream的数据数据源

2、在输入DStream上进行transformations和输出操作。

3、调用streamingContext.start()来启动上面的程序处理。

4、调用streamingContext.awaitTermination来等待程序结束(人为停止或程序出错)

5、可以通过streamingContext.stop()来人工停止程序。

需要记住的是:

1、一旦context已经开始(streamingContext.start()),新的流计算就不能被创建并加入到程序中了。

2、context被stop后,不能restart。

3、一个jvm中同一时刻只有一个streamingContext是active的。

4、streamingContext的stop函数也会把SparkContext给stop掉。如果仅仅想把streamingContext停掉,那么就要给stop函数传入stopSparkContext=false的参数。

5、SparkContext可以重复利用来创建多个SteamingContext,只要之前的SteamingContext先被stop(只停streamingContext不停SparkContext)。

3.3 离散数据流(DStreams)

DStreams代表源源不断的RDD的集合,每个rdd是一段时间内的流数据,如下图所示(参考上面的例子,一秒一个RDD):

在这些RDD上面可以做各种的高层函数处理,如下:

3.4 输入DStreams和接收器

输入DStream是从数据源得到的一个数据流抽象概念(源源不断的RDD集合),在上面的例子中,lines就是一个从tcp socket接收数据的一个输入DStream。每个一输入DStream(除文件系统),都与一个Receiver接收器密切相关,这个接收器负责从数据源接收数据到内存等待处理。

spark streaming提供两种类型的内建流数据源:

1、基础类型数据源:直接在SteamingContext API中就支持的(文件系统和socket连接),无需导入别的maven依赖。

2、高级数据源:类似kafka、flume、kinesis等,需要导入额外的依赖包才能够使用。

需要注意的是:如果你想在你的流应用中并行接收多个流式数据,你可以创建多个输入DStream。这样会同时创建多个接收器来并行接收多个数据流的数据。但是要注意,一个spark worker/executor是一个长时间运行的任务,会占用一个你的应用分配的内核,所以你需要分配足够的内核(或者线程(本地运行的话))给你的流应用,不但用于接收器接收流数据,还要用户处理接收到的数据。

记住:

1、如果你本地运行spark streaming程序,不要设置master为“local”或“local[1]”(代表只会有一个本地线程来跑你的程序)。这时候如果你使用基于接收器的输入DStream(例如sockets,kafka、flume等),这个唯一的线程就会被接收器占用,那么就没有空余线程去处理接收到的数据了。所以本地运行,需要设置master为“local[n]”,n要大于接收器的数量。

2、如果集群模式,应用分配的内核数量必须多余接收器的数量,否则,系统就只能接收数据,但不能去处理它了。

3.4.1 基础数据源

上面的例子我们已经看过了TCP socket的基础数据源的使用方式,现在看一下文件系统基础数据源的使用。

文件系统:

能够从兼容hdfs api的文件系统(hdfs、s3、nfs等)中读取文件,可以通��StreamingContext.fileSteam[KeyClass,ValueClass,InputFormatClass]来创建输入DStream。因为文件系统数据源不需要接收器,所以不需要为接收器分配内核来接收数据。api如下:

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

对于文本文件,可以使用如下api:

streamingContext.textFileStream(dataDirectory)

上面api的文件目录如何监听?

spark streaming会监听dataDirectory目录,并处理在这个目录中创建的任何一个文件。

1、例如“hdfs://namenode:8040/logs/”这种简单的目录会被监控,这个目录中的文件会被发现并处理。(经验证,只会监控logs下的文件,子目录里不会监控处理)

2、可以使用通配符匹配,例如:“hdfs://namenode:8040/logs/2017/*”匹配上的目录都会被处理。这个是目录的匹配模式,而不是文件的。(经验证,是只会匹配目录,文件忽略)

3、所有的文件必须是同一数据类型。(这个很好理解,fileSteam函数需指定InputFormatClass,所以如果不同类型,就不对了)

4、文件是根据它的修改时间,而不是创建时间来监控处理的。(比如一个文件cp到监控目录,会被处理,然后你又去修改了这个文件,那么这些修改是被忽略的)

5、一旦文件被处理了,那么修改文件将不会触发重新读取,也就是说更新会被忽略。(比如一个文件cp到监控目录,会被处理,然后你又去修改了这个文件,那么这些修改是被忽略的)

5、一个目录下的文件越多,将会花费更多的时间来扫描更新,即使没有文件被修改。(这个也很好理解,监控目录的时候,是需要一个个文件去看它的修改时间的,所以即使文件没有更新,也是要一个个扫描的)

6、如果使用了通配符,那么如果你修改了整个目录的名称,而这个名称恰好能够被匹配到,这个目录就会被监控处理。目录中的文件,只有修改时间在目前的窗口内才会被读取处理。

7、可以调用FileSystem.setTimes()函数来修改文件时间戳,然后稍后的时间窗口就可以被处理了,尽管这个文件没有改变。

 

使用类似hdfs这种文件系统,比如你代码中写数据到文件的时候,一旦output stream创建了,那么你的文件的修改时间也就是output stream的创建时间,但你的数据会在这个时刻之后被写入(有可能时间较长),那么根据目录监控只能根据文件修改时间的规则,在你创建output stream的时间窗口内的数据会被处理(也就是上面说的1秒),1秒之后的数据会被忽略。

所以你可以先在别的目录中创建文件并写入数据,然后等写完之后拷贝或者重命名到监控目录就可以了。

相比之下,类似Amazon S3和Azure这种存储系统,一般会保证数据写完才会确定修改时间。而且,如果要重命名文件,那么修改时间就是这个重命名的时间(本地文件系统重命名文件是不会被监控到的)。

 

可以自定义receiver来接收数据

 

3.4.2 高级数据源

类似kafka、flume这种,需要导入额外的依赖包。所以如果spark shell是不支持这些数据源的,如果你想在spark shell中使用,那么就需要下载这些maven依赖包,然后把这些包加到classpath中。

3.4.3 自定义数据源

你也可以自定义数据源,你所要做的就是自定义自己的receiver,从自定义数据源中能够接收数据即可。

3.4.4 接收器可靠性

根据消息可靠性,有两种数据源,像kafka、flume这种提供消息确认机制的数据源,在你接收到数据后需要确认消息已送达,这样就会保证数据没有丢失:

1、可靠的接收器:当收到数据并容错存储后,接收器需要发送消息已收到的确认给数据源。

2、不可靠的接收器:接收到数据后不会发确认消息,不支持消息确认或者支持但不想确认的,都可以使用这种方式。

3.5 DStreams转换操作

和RDDs类似,支持普通Rdd的很多转换操作,详见官方文档。下面是一些比较特殊的操作:

3.5.1 UpdateStateByKey操作

updateStateByKey操作允许你使用流数据的新数据来维持一个持续不断的状态更新。步骤如下:

1.定义一个状态(可任意数据类型)

2.定义状态更新函数(指定怎么用之前的状态和新到的数值来更新这个状态)

每个batch(其实就是一个RDD)里,更新函数会被应用到所有存在的key上,不管在这个批次里这个key是否有新数据。如果更新函数返回None,那么这个key的键值对会被移除。

让我们验证下,统计文本数据中单词个数,状态count是integer型,更新函数定义如下(newValues是流数据的一个个新到数据,runningCount是新到数据之前的状态,返回值是更新后的状态):

  1. def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
  2. val newCount = ... // add the new values with the previous running count to get the new count
  3. Some(newCount)
  4. }

然后下面就是使用这个跟新函数:

val runningCounts = pairs.updateStateByKey[Int](updateFunction _)(pairs是之前例子中的DStream,其实内部就是(word,1)类型的RDD)

然后newValues就是(word,1),word就是key,就会根据单词进行加1计数。

需要注意的是,使用updateStateKey需要配置checkpoint目录(因为要维持一个源源不断的状态计数,所以如果程序出现问题,这个状态就没办法维持了,丢失了之前的状态)

3.5.2 transform操作

transform操作(还有它的变种transformWith操作),允许你在DStream上应用任意的RDD-to-RDD转换函数。如果DStream api中没有暴露一些RDD的操作函数,这里就可以用这个方法进行处理。例如,DStream的api中没有把数据流的每个批次和另一个数据集关联的功能。但这里你就可以用transform进行了。例如,

收藏 推荐 打印 | 阅读: