尽管SparkStreaming曾经进行更新，Spark的重点也放到了 Structured Streaming ，但因为Spark版本过低或者其余技术选型问题，可能还是会抉择SparkStreaming。
SparkStreaming对于工夫窗口，事件工夫尽管撑持较少，但还是能够满足局部的实时计算场景的，SparkStreaming材料较多，这里也做一个简略介绍。

一． 什么是Spark Streaming

Spark Streaming在过后是为了与过后的Apache Storm竞争，也让Spark能够用于流式数据的解决。依据其官网文档介绍，Spark Streaming有高吞吐量和容错能力强等特点。Spark Streaming反对的数据输出源很多，例如：Kafka、Flume、Twitter、ZeroMQ和简略的TCP套接字等等。数据输出后能够用Spark的高度形象原语如：map、reduce、join、window等进行运算。而后果也能保留在很多中央，如HDFS，数据库等。另外Spark Streaming也能和MLlib（机器学习）以及Graphx完满交融。
当然Storm目前曾经慢慢淡出，Flink开始大放异彩。

Spark与Storm的比照

二、SparkStreaming入门

Spark Streaming 是 Spark Core API 的扩大，它反对弹性的，高吞吐的，容错的实时数据流的解决。数据能够通过多种数据源获取，例如 Kafka，Flume，Kinesis 以及 TCP sockets，也能够通过例如 map，reduce，join，window 等的高级函数组成的简单算法解决。最终，解决后的数据能够输入到文件系统，数据库以及实时仪表盘中。事实上，你还能够在 data streams（数据流）上应用 [机器学习] 以及 [图计算] 算法。
在外部，它工作原理如下，Spark Streaming 接管实时输出数据流并将数据切分成多个 batch（批）数据，而后由 Spark 引擎解决它们以生成最终的 stream of results in batches（分批流后果）。

Spark Streaming 提供了一个名为 discretized stream 或 DStream 的高级形象，它代表一个间断的数据流。DStream 能够从数据源的输出数据流创立，例如 Kafka，Flume 以及 Kinesis，或者在其余 DStream 上进行高层次的操作以创立。在外部，一个 DStream 是通过一系列的 [RDDs] 来示意。

本指南通知你如何应用 DStream 来编写一个 Spark Streaming 程序。你能够应用 Scala，Java 或者 Python（Spark 1.2 版本后引进）来编写 Spark Streaming 程序。

在idea中新建maven我的项目

引入依赖

<dependency>            <groupId>org.apache.spark</groupId>            <artifactId>spark-streaming_2.11</artifactId>            <version>2.4.4</version>        </dependency>

Project Structure —— Global Libraries —— 把scala 增加到 add module

新建Scala Class

import org.apache.log4j.{Level, Logger}import org.apache.spark.SparkConfimport org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}import org.apache.spark.streaming.{Seconds, StreamingContext}object Demo {  //屏蔽日志  Logger.getLogger("org.apache")setLevel(Level.WARN)  def main(args: Array[String]): Unit = {    //local会有问题  起码两个线程  一个拿数据 一个计算    //val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local")    val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[2]")    //工夫距离    val ssc = new StreamingContext(conf,Seconds(1))    //接收数据 解决    //socket  demo    val value: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)    val words: DStream[String] = value.flatMap(_.split(" "))    val wordsTuple: DStream[(String, Int)] = words.map((_, 1))    val wordcount: DStream[(String, Int)] = wordsTuple.reduceByKey(_ + _)    //触发action    wordcount.print()    ssc.start()    //放弃流的运行  期待程序被终止    ssc.awaitTermination()  }}

测试

下载一个win10 用的netcat

https://eternallybored.org/mi...

下载netcat 1.12

解压 在目录下启动cmd

输出

nc  -L -p 9999

开始输出单词 在idea中验证接管

原理

初始化StreamingContext

为了初始化一个 Spark Streaming 程序，一个 StreamingContext 对象必须要被创立进去，它是所有的 Spark Streaming 性能的主入口点。

import org.apache.spark._import org.apache.spark.streaming._val conf = new SparkConf().setAppName(appName).setMaster(master)val ssc = new StreamingContext(conf, Seconds(1))

appName 参数是展现在集群 UI 界面上的应用程序的名称

master 是local 或者spark集群的url（mesos yarn）

本地测试能够用local[*] 留神要多于两个线程

Second（1）定义的是batch interval 批处理距离 就是距离多久去拿一次数据

在定义一个 context 之后,您必须执行以下操作。

1. 通过创立输出 DStreams 来定义输出源。
2. 通过利用转换和输入操作 DStreams 定义流计算（streaming computations）。
3. 开始接管输出并且应用 streamingContext.start() 来解决数据。
4. 应用 streamingContext.awaitTermination() 期待解决被终止（手动或者因为任何谬误）。
5. 应用 streamingContext.stop() 来手动的进行解决。

须要记住的几点:

• 一旦一个 context 曾经启动，将不会有新的数据流的计算能够被创立或者增加到它。
• 一旦一个 context 曾经进行，它不会被重新启动。
• 同一时间外在 JVM 中只有一个 StreamingContext 能够被激活。
• 在 StreamingContext 上的 stop() 同样也进行了 SparkContext。为了只进行 StreamingContext，设置 stop() 的可选参数，名叫 stopSparkContext 为 false。
• 一个 SparkContext 就能够被重用以创立多个 StreamingContexts，只有前一个 StreamingContext 在下一个StreamingContext 被创立之前进行（不进行 SparkContext）。

Discretized Stream or DStream

Discretized Stream or DStream 是 Spark Streaming 提供的根本形象。它代表了一个间断的数据流。可能是数据源接管的流，也可能是转换后的流。

DStream就是多个和工夫相干的一系列间断RDD的汇合，比方本例就是距离一秒的一堆RDD的汇合

DStream也是有依赖关系的

flatMap 操作也是间接作用在DStream上的，就和作用于RDD一样 这样很好了解

咱们先来看数据源接管的流 这种叫做Input DStreams 他会通过Receivers接收器去不同的数据源接收数据。

Spark Streaming内置了两种数据源：

• 根底的数据源：比方方才用的socket接管 还有file systems
• 高级的数据源：比方kafka 还有flume kinesis等等

留神本地运行时，不要用local或者local[1],一个线程不够。放到集群上时调配给SparkStreaming的核数必须大于接收器的数量，留一个核去解决数据。

咱们也能够自定义数据源，那咱们就须要本人开发一个接收器。

Transformations

在咱们接管到Dstreams之后能够进行转换操作，常见转换如下：

这里咱们特地介绍一下updateStateByKey

咱们如果须要对历史数据进行统计，可能须要去kafka里拿一下之前留存的数据，也能够用updateStateByKey这个办法。

//保留状态  聚合雷同的单词    val  wordcount = wordsTuple.updateStateByKey[Int](      //updateFunction _      (newValues: Seq[Int], runningCount: Option[Int])=> {        val newCount = Some(newValues.sum + runningCount.getOrElse(0))        newCount      }    )

比方方才的单词计数，咱们只能统计每一次发过来的音讯，然而如果心愿统计屡次音讯就须要用到这个，咱们要指定一个checkpoint，就是从哪开始算。

//减少成员变量val checkpointDir = "./ckp"//在办法中退出checkpointssc.checkpoint(checkpointDir)    val value: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)    value.checkpoint(Seconds(4))//官网倡议批次工夫的1-5倍

这时候咱们建设StreamingContext的办法就要扭转了 咱们把方才的创立过程提取成办法。

def creatingFunc():StreamingContext = {    val conf = new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")    val ssc = new StreamingContext(conf, Seconds(1))    ssc.checkpoint(checkpointDir)    val value: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)    value.checkpoint(Seconds(4))//官网倡议批次工夫的1-5倍    val words: DStream[String] = value.flatMap(_.split(" "))    val wordsTuple: DStream[(String, Int)] = words.map((_, 1))    //保留状态  聚合雷同的单词    val  wordcount = wordsTuple.updateStateByKey[Int](      //updateFunction _      (newValues: Seq[Int], runningCount: Option[Int])=> {        val newCount = Some(newValues.sum + runningCount.getOrElse(0))        newCount      }    )    //触发action    wordcount.print()    ssc  }

在mian函数中批改为：

def main(args: Array[String]): Unit = {      val ssc = StreamingContext.getOrCreate(checkpointDir,creatingFunc _)      ssc.start()      //放弃流的运行  期待程序被终止      ssc.awaitTermination()}

这样就是，如果有checkpoint，程序会在checkpoint中把程序加载回来（程序被保留为二进制），没有checkpoint的话才会创立。

将目录下的checkpoint删除，就能够将状态删除。

生产中updateStateByKey因为会将数据备份要谨慎应用，能够思考用hbase，redis等做代替。或者借助kafka做聚合解决。

//如果不必updatestateByKey  能够思考redis    wordsTuple.foreachRDD(rdd => {      rdd.foreachPartition(i =>        {          //redis        }      )    })

窗口操作

Spark Streaming 也反对 _windowed computations（窗口计算），它容许你在数据的一个滑动窗口上利用 transformation（转换）。

如上图显示，窗口在源 DStream 上 _slides（滑动），任何一个窗口操作都须要指定两个参数：

• window length（窗口长度） - 窗口的持续时间。
• sliding interval（滑动距离） - 执行窗口操作的距离。

比方计算过来30秒的词频：

val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

一些罕用的窗口操作如下所示，这些操作都须要用到上文提到的两个参数 - windowLength（窗口长度） 和 _slideInterval（滑动的工夫距离）_。

Join操作

在 Spark Streaming 中能够执行不同类型的 join

val stream1: DStream[String, String] = ...val stream2: DStream[String, String] = ...val joinedStream = stream1.join(stream2)//也能够用窗口val windowedStream1 = stream1.window(Seconds(20))val windowedStream2 = stream2.window(Minutes(1))val joinedStream = windowedStream1.join(windowedStream2)

DStreams输入操作

输入操作容许将 DStream 的数据推送到内部零碎，如数据库或文件系统。

会触发所有变换的执行，相似RDD的action操作。有如下操作：

foreachRDD设计模式应用

dstream.foreachRDD容许将数据发送到内部零碎。

但咱们不要每次都创立一个连贯，解决方案如下：

缩小开销，分区摊派开销

dstream.foreachRDD { rdd =>  rdd.foreachPartition { partitionOfRecords =>    val connection = createNewConnection()    partitionOfRecords.foreach(record => connection.send(record))    connection.close()  }}

更好的做法是用动态资源池：

dstream.foreachRDD { rdd =>  rdd.foreachPartition { partitionOfRecords =>    // ConnectionPool is a static, lazily initialized pool of connections    val connection = ConnectionPool.getConnection()    partitionOfRecords.foreach(record => connection.send(record))    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse  }}

连贯Kafka

Apache Kafka是一个高性能的音讯零碎，由Scala 写成。是由Apache 软件基金会开发的一个开源音讯零碎我的项目。

Kafka 最后是由LinkedIn 开发，并于2011 年初开源。2012 年10 月从Apache Incubator 毕业。该项目标指标是为解决实时数据提供一个对立、高通量、低期待（低延时）的平台。

更多kafka相干请查看Kafka入门宝典（具体截图版）

Spark Streaming 2.4.4兼容 kafka 0.10.0 或者更高的版本

Spark Streaming在2.3.0版本之前是提供了对kafka 0.8 和 0.10的反对的 ，不过在2.3.0当前对0.8的反对勾销了。

Note: Kafka 0.8 support is deprecated as of Spark 2.3.0.

Receiver

这里简略介绍一下对kafka0.8的一种反对形式：基于Receiver

依赖：

groupId = org.apache.spark artifactId = spark-streaming-kafka-0-8_2.12 version = 2.4.4

import org.apache.spark.streaming.kafka._ val kafkaStream = KafkaUtils.createStream(streamingContext,     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

这种状况 程序停掉数据会失落，为了不失落本人又写了一份，这种是很多余的。

因为采纳了kafka高阶api，偏移量offset不可控。

Direct

Kafka 0.10.0版本当前,采纳了更好的一种Direct形式，这种咱们须要本人保护偏移量offset。

直连形式 并行度会更高 生产环境用的最多，0.8版本须要在zk或者redis等中央本人保护偏移量。咱们应用0.10以上版本反对本人设置偏移量，咱们只须要本人将偏移量写回kafka就能够。

依赖

groupId = org.apache.sparkartifactId = spark-streaming-kafka-0-10_2.12version = 2.4.4

kafka 0.10当前 能够将offset写回kafka 咱们不须要本人保护offset了，具体代码如下：

val conf = new SparkConf().setAppName("KafkaStreaming").setMaster("local[*]")    val ssc = new StreamingContext(conf,Seconds(2))    val kafkaParams = Map[String, Object](      "bootstrap.servers" -> "localhost:9092,anotherhost:9092",      "key.deserializer" -> classOf[StringDeserializer],      "value.deserializer" -> classOf[StringDeserializer],      "group.id" -> "use_a_separate_group_id_for_each_stream",      //latest  none   earliest      "auto.offset.reset" -> "earliest",      //主动提交偏移量 false      "enable.auto.commit" -> (false: java.lang.Boolean)    )    //val topics = Array("topicA", "topicB")    val topics = Array("test_topic")    val stream = KafkaUtils.createDirectStream[String, String](      ssc,      // 与kafka broker不在一个节点上  用不同策略      //在一个节点用 PreferBrokers策略  很少见      LocationStrategies.PreferConsistent,      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)    )    stream.foreachRDD(rdd => {      //一般的RDD不能强转HasOffsetRanges   但kafkaRDD有 with这个个性 能够强转      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges      //解决数据 计算逻辑      rdd.foreachPartition { iter =>        //一次解决一个分区的数据  获取这个分区的偏移量        //计算完当前批改偏移量  要开启事务 相似数据库 connection -> conn.setAutoCommit(false) 各种操作  conn.commit(); conn.rollback()        //获取偏移量  如果要本人记录的话这个        //val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)        //println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")        //解决数据         iter.foreach(println)      }      //kafka 0.10新个性  解决完数据后  将偏移量写回kafka      // some time later, after outputs have completed      //kafka有一个非凡的topic  保留偏移量      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)    })

更多Flink，Kafka，Spark等相干技术博文，科技资讯，欢送关注实时流式计算 公众号后盾回复 “电子书” 下载300页Flink实战电子书