一、流式计算简介

1.1. SparkStreaming简介

编辑1.2. SparkStreaming基本工作原理

1.3. Storm V.S. SparkStreaming V.S. Flink

二、SparkStreaming实时处理入门

2.1. 工程创建

2.2. 入口类StreamingContext

2.3. 实例练习

三、SparkStreaming与Kafka整合

3.1. 整合简述

3.2. 两种整合方式

3.3. 实操(Direct 方式)

四、SparkStreaming常见transformation算子

五、window

六、SparkSQL和SparkStreaming的整合案例

一、流式计算简介

大数据中的计算的数据通常可分为以下三大类：
1、一种是离线数据(T+1,周、月、季度、年等指标)

2、一种是实时数据(一条数据触发一次计算、较短时间触发一次计算、最近几秒、最近几分钟、最近几十分钟)

3、一种是准实时(交互式)（较短时间触发一次计算、最近几分钟、最近几十分钟）

1.1. SparkStreaming简介
        SparkStreaming，和SparkSQL一样，也是Spark生态栈中非常重要的一个模块，主要是用来进行流式计算的框架。流式计算框架，从计算的延迟上面，又可以分为纯实时流式计算和准实时流式计算，SparkStreaming是属于的准实时计算框架。

        纯实时的计算，指的是来一条记录(event事件)，启动一次计算的作业；离线计算，指的是每次计算一个非常大的一批(比如几百G，好几个T)数据；准实时计算介于纯实时和离线计算之间的一种计算方式，不是每一条记录就计算一次，比离线计算数据量小的多，计算数据量可表示为Micro-batch（微小的批次）。

        SparkStreaming是SparkCore的api的一种扩展，使用DStream(discretized stream or DStream)作为数据模型，基于内存处理连续的数据流，本质上还是RDD的基于内存的计算。

        DStream，本质上是RDD的序列。SparkStreaming的处理流程可以归纳为下图：

1.2. SparkStreaming基本工作原理
        接收实时输入数据流，然后将数据拆分成多个batch，比如每收集1秒的数据封装为一个batch，然后将每个batch交给Spark的计算引擎进行处理，最后会生产出一个结果数据流，其中的数据，也是由一个一个的batch所组成的。

        Spark Streaming提供了一种高级的抽象，叫做DStream，英文全称为Discretized Stream，中文翻译为“离散流”，它代表了一个持续不断的数据流。DStream可以通过输入数据源来创建，比如Kafka、Flume、ZMQ和Kinesis；也可以通过对其他DStream应用高阶函数来创建，比如map、reduce、join、window。
        DStream的内部，其实一系列持续不断产生的RDD。RDD是Spark Core的核心抽象，即，分布式式弹性数据集。DStream中的每个RDD都包含了一个时间段内的数据。

         对DStream应用的算子，比如map，其实在底层会被翻译为对DStream中每个RDD的操作。比如对一个DStream执行一个map操作，会产生一个新的DStream。但是，在底层，其实其原理为，对输入DStream中每个时间段的RDD，都应用一遍map操作，然后生成的新的RDD，即作为新的DStream中的那个时间段的一个RDD。底层的RDD的transformation操作，还是由Spark Core的计算引擎来实现的。Spark Streaming对Spark Core进行了一层封装，隐藏了细节，然后对开发人员提供了方便易用的高层次的API。

1.3. Storm V.S. SparkStreaming V.S. Flink

二、SparkStreaming实时处理入门

2.1. 工程创建

在pom.xml文件中导入SparkStreaming的maven依赖

  
        
            org.apache.spark
            spark-streaming_2.12
            3.1.2

在com package下创建一个新的package named sparkStreaming_example。

2.2. 入口类StreamingContext

SparkStreaming中的入口类，称之为StreamingContext,但是底层还是得需要依赖SparkContext。

object SparkStreamingTest {
    def main(args: Array[String]): Unit = {
        
        val conf = new SparkConf()
                    .setAppName("SparkStreamingTest")
                    .setMaster("local[*]")
        val duration = Seconds(2)
        val ssc = new StreamingContext(conf, duration)

        //业务
        
        
        //为了执行的流式计算，必须要调用start来启动
        ssc.start()
        
        ssc.awaitTermination()
    }
}

2.3. 实例练习

SparkStreaming是一个流式计算的计算引擎，那么就模拟一个对流式数据进行单词统计。

这里我们需要用到netcat，Netcat（简称nc）是一款强大的命令行网络工具，用来在两台机器之间建立TCP/UDP连接，并通过标准的输入输出进行数据的读写。

可以使用nc命令查看系统中是否有安装Netcat，如果没有安装，输入以下命令安装

yum install -y nc

使用场景:

1)与服务器建立连接

使用命令: nc 要建立连接的服务器地址连接端口

功能：如 nc www.baidu.com 80 可与百度服务器建立TCP连接

连接后可向百度服务器发送GET请求，模拟浏览器访问百度首页，输入GET / HTTP/1.0并按两次回车

2）使用命令：nc -lk port号
说明： -l为listen，监听port端口

-k 强制 nc 待命链接。当客户端从服务端断开连接后，过一段时间服务端也会停止监听。

但通过选项 -k 我们可以强制服务器保持连接并继续监听端口。
功能：启动了一个可以发送socket文本流的服务器的端口，该端口等待客户端来连接，端口号为port

退出重新进入时如出现

Ncat: bind to :::6666: Address already in use. QUITTING.
下载net-tools

yum install net-tools

通过命令netstat -nap | grep 6666找到占用进程

再通过命令kill -9 进程号杀死进程，重新监听就可以了

代码实现

package com.sparkStreaming_example

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}

object StreamingWordCount {
  def main(args: Array[String]): Unit = {
    //1、获取SparkStreaming上下文环境
    val conf = new SparkConf().setAppName("streaming-wc").setMaster("local[*]")
    //Durations.seconds(5)  5秒处理一次数据
    val ssc: StreamingContext = new StreamingContext(conf, Durations.seconds(5))
    //2、获取数据源
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("user1", 6666)
    //3、对获取的数据进行加工处理
    val sumed: DStream[(String, Int)] = ds.filter(_.length > 0)
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
    //4、将结果数据进行持久化
    sumed.print()
    //5、开启流程序并等待结束信号
    ssc.start()
    ssc.awaitTermination() //等待结束信号
  }
}

三、SparkStreaming与Kafka整合

3.1. 整合简述
kafka是做消息的缓存，数据和业务隔离操作的消息队列，而sparkstreaming是一款准实时流式计算框架，所以二者的整合，是大势所趋。

二者的整合，有主要的两大版本。

kafka作为一个实时的分布式消息队列，实时的生产和消费消息，在实际开发中Spark Streaming经常会结合Kafka来处理实时数据。Spark Streaming 与 kafka整合需要引入spark-streaming-kafka.jar，该jar根据kafka版本有2个分支，分别是spark-streaming-kafka-0-8 和 spark-streaming-kafka-0-10。jar包分支选择原则：

0.10.0>kafka版本>=0.8.2.1，选择 08 接口
kafka版本>=0.10.0，选择 010 接口

sparkStreaming和Kafka整合一般两种方式：Receiver方式和Direct方式

3.2. 两种整合方式
Receiver方式

        Receiver方式基于kafka的高级消费者API实现（高级优点：高级API写起来简单；不需要去自行去管理offset，系统通过zookeeper自行管理；不需要管理分区，副本等情况，系统自动管理；消费者断线会自动根据上一次记录在 zookeeper中的offset去接着获取数据；高级缺点：不能自行控制 offset；不能细化控制如分区、副本、zk 等）。Receiver从kafka接收数据，存储在Executor中，Spark Streaming 定时生成任务来处理数据。

       默认配置的情况，Receiver失败时有可能丢失数据。如果要保证数据的可靠性，需要开启预写式日志，简称WAL（Write Ahead Logs，Spark1.2引入），只有接收到的数据被持久化之后才会去更新Kafka中的消费位移。接收到的数据和WAL存储位置信息被可靠地存储，如果期间出现故障，这些信息被用来从错误中恢复，并继续处理数据。

Direct的方式

        Direct方式从Spark1.3开始引入的，通过 KafkaUtils.createDirectStream 方法创建一个DStream对象，Direct方式的结构如下图所示：

Direct 方式特点如下：

Offset 可自定义
使用kafka低阶API
底层实现为KafkaRDD

        该方式中Kafka的一个分区与Spark RDD对应，通过定期扫描所订阅Kafka每个主题的每个分区的最新偏移量以确定当前批处理数据偏移范围。与Receiver方式相比，Direct方式不需要维护一份WAL数据，由Spark Streaming程序自己控制位移的处理，通常通过检查点机制处理消费位移，这样可以保证Kafka中的数据只会被Spark拉取一次。

3.3. 实操(Direct 方式)

引入依赖


    org.apache.spark
    spark-streaming-kafka-0-10_2.12
    3.1.2

模拟kafka生产数据

运行程序前先启动kafka的消费者客户端

/usr/local/kafka-2.4.1/bin/kafka-console-consumer.sh 
--bootstrap-server user1:9092 
--topic test 
--from-beginning

kafka生产数据代码

package com.sparkStreaming_example

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import java.util.{Properties, Random}


object DataToKafka {
  def main(args: Array[String]): Unit = {
    val prop = new Properties()
    //提供Kafka服务器信息
    prop.put("bootstrap.servers", "user1:9092")
    //指定响应的方式
    prop.put("acks", "all")
    //请求失败重试的次数
    prop.put("retries", "3")
    //指定key的序列化方式，key是用于存放数据对应的offset
    prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value的序列化方式
    prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //创建producer对象
    val producer = new KafkaProducer[String, String](prop)
    //提供一个数组，数组中数据
    val arr = Array(
      "hello zhangsan",
      "hello lisi",
      "hello wangwu",
      "hello zhaoliu",
      "hello sunqi",
      "hello zhouba",
    )
    //提供一个随机数，随机获取数组中数据向kafka中进行发送存储
    val r = new Random()
    while (true) {
      val message = arr(r.nextInt(arr.length))
      producer.send(new ProducerRecord[String, String]("test", message))
      //      Thread.sleep(r.nextInt(1000))
      Thread.sleep(1000)
    }
  }
}

程序运行后，消费者客户端消费数据

实时消费kafka数据

package com.sparkStreaming_example

import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

object StreamingWithKafka {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf对象
    val conf = new SparkConf()
      .setAppName("SparkStreamingWithKafka")
      .setMaster("local[*]")
    //2.提供批次时间
    val time = Seconds(3)
    //3.提供StreamingContext对象
    val sc = new StreamingContext(conf,time)
    //4.提供Kafka配置参数
    val kafka = Map[String,Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG->"user1:9092",
      ConsumerConfig.GROUP_ID_CONFIG->"user",
      "key.deserializer"->"org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer"->"org.apache.kafka.common.serialization.StringDeserializer",
    )
    //5.读取Kafka中数据信息生成DStream
    val value = KafkaUtils.createDirectStream(sc,
      //本地化策略：将Kafka的分区数据均匀的分配到各个执行Executor中
      LocationStrategies.PreferConsistent,
      //表示要从使用kafka进行消费【offset谁来管理，从那个位置开始消费数据】
      ConsumerStrategies.Subscribe[String, String](Set("test"), kafka)
    )
    //6.将每条消息kv获取出来
    val line: DStream[String] = value.map(record => record.value())
    //7.开始计算操作
    line.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
    //8.开始任务
    sc.start()
    sc.awaitTermination()
  }
}

运行程序后不断从kafka读取数据

说明

简化的并行性：不需要创建多个输入Kafka流并将其合并。使用directStream，Spark Streaming将创建与使用Kafka分区一样多的RDD分区，这些分区将全部从Kafka并行读取数据。所以在Kafka和RDD分区之间有一对一的映射关系。
效率：在第一种方法中实现零数据丢失需要将数据存储在预写日志中，这会进一步复制数据。这实际上是效率低下的，因为数据被有效地复制了两次:一次是Kafka，另一次是由预先写入日志（WriteAhead Log）复制。这个第二种方法消除了这个问题，因为没有接收器，因此不需要预先写入日志。只要Kafka数据保留时间足够长。
正好一次（Exactly-once）的语义：第一种方法使用Kafka的高级API来在Zookeeper中存储消耗的偏移量。传统上这是从Kafka消费数据的方式。虽然这种方法（结合提前写入日志）可以确保零数据丢失（即至少一次语义），但是在某些失败情况下，有一些记录可能会消费两次。发生这种情况是因为Spark Streaming可靠接收到的数据与Zookeeper跟踪的偏移之间的不一致。因此，在第二种方法中，我们使用不使用Zookeeper的简单Kafka API。在其检查点内，Spark Streaming跟踪偏移量。这消除了Spark Streaming和Zookeeper/Kafka之间的不一致，因此Spark Streaming每次记录都会在发生故障的情况下有效地收到一次。为了实现输出结果的一次语义，将数据保存到外部数据存储区的输出操作必须是幂等的，或者是保存结果和偏移量的原子事务。

四、SparkStreaming常见transformation算子

cogroup

cogroup就是groupByKey的另外一种变体，groupByKey是操作一个K-V键值对，而cogroup一次操作两个，有点像join，不同之处在于返回值结果：

#cogroup()
val ds1:DStream[(K, V)]
val ds2:DStream[(K, w)]
val cg:DStream[(K, (Iterable[V], Iterable[W]))] = ds1.cogroup(ds2)
#groupByKey()
val gp:DStream[(K, Iterable[V])] = ds1.groupByKey()

updateStateByKey

updateStateByKey(func) 根据于key的前置状态和key的新值，对key进行更新，返回一个新状态的Dstream。
功能：统计截止到目前为止key的状态。
在这个操作中需要两个数据，一个是key的前置状态，一个是key的新增(当前批次的数据)；还有历史数据(前置状态)得需要存储在磁盘，不应该保存在内存中。
同时key的前置状态可能有可能没有。

案例实现——wordcount

package com.sparkStreaming_example

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}


object UpdateStateByKeyTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("WordCountUpdateStateByKey")
      .setMaster("local[*]")
    val duration = Seconds(5)
    val ssc = new StreamingContext(conf, duration)
    ssc.checkpoint("C:\Users\80621\Desktop\data\streamData\chk")
    val lines:DStream[String] = ssc.socketTextStream("user1", 6666)

    val pairs:DStream[(String, Int)] = lines.flatMap(_.split("\s+")).map((_, 1))

    val usb:DStream[(String, Int)] = pairs.updateStateByKey(updateFunc)
    usb.print()
    ssc.start()
    ssc.awaitTermination()
  }

  
  def updateFunc(seq: Seq[Int], option: Option[Int]): Option[Int] = {
    println("option:" + option +  "> seq: " + seq.mkString("[", ",", "]"))
    
    Option(seq.sum + option.getOrElse(0))
  }
}

在端口6666中不断输入，程序累计输出

五、window

滚动窗口(左闭右开)+滑动窗口

window操作就是窗口函数。Spark Streaming提供了滑动窗口操作的支持，从而让我们可以对一个滑动窗口内的数据执行计算操作。每次掉落在窗口内的RDD的数据，会被聚合起来执行计算操作，然后生成的RDD，会作为window DStream的一个RDD。比如下图中，就是对每三秒钟的数据执行一次滑动窗口计算，这3秒内的3个RDD会被聚合起来进行处理，然后过了两秒钟，又会对最近三秒内的数据执行滑动窗口计算。所以每个滑动窗口操作，都必须指定两个参数，窗口长度以及滑动间隔，而且这两个参数值都必须是batch间隔的整数倍。

说明：

红色的矩形就是一个窗口，窗口hold的是一段时间内的数据流。
这里面每一个time都是时间单元，在官方的例子中，每隔window size是3 time unit, 而且每隔2个单位时间，窗口会slide一次。
所以基于窗口的操作，需要指定2个参数：
window length - The duration of the window (3 in the figure)
slide interval - The interval at which the window-based operation is performed (2 in the figure).
窗口大小，个人感觉是一段时间内数据的容器。
滑动间隔，就是我们可以理解的cron表达式

案例实现

package com.sparkStreaming_example

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}


object SlideWindow {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("WordCountUpdateStateByKey")
      .setMaster("local[*]")
    val batchInterval = 2
    val duration = Seconds(batchInterval)
    val ssc = new StreamingContext(conf, duration)
    val lines:DStream[String] = ssc.socketTextStream("user1", 6666)
    val pairs:DStream[(String, Int)] = lines.flatMap(_.split("\s+")).map((_, 1))

    val ret:DStream[(String, Int)] = pairs.reduceByKeyAndWindow(_+_,
      windowDuration = Seconds(batchInterval * 3),  //窗口长度3s
      slideDuration = Seconds(batchInterval * 2))  //滑动间隔2s

    ret.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

六、SparkSQL和SparkStreaming的整合案例

Spark最强大的地方在于，可以与Spark Core、Spark SQL整合使用，之前已经通过transform、foreachRDD等算子看到，如何将DStream中的RDD使用Spark Core执行批处理操作。现在就来看看，如何将DStream中的RDD与Spark SQL结合起来使用。
案例：top3的商品排序：最新的top3

这里就是基于updatestateByKey，统计截止到目前为止的不同品类下的商品销量top3。

DS-->RDD-->DataFrame-->注册成表-->对表进行sql-->展示结果

代码

package com.sparkStreaming_example

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

object sparkSQLWithStreaming {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("StreamingWithSQL")
      .setMaster("local[*]")
    val batchInterval = 3
    val duration = Seconds(batchInterval)
    val spark = SparkSession.builder()
      .config(conf)
      .getOrCreate()
    //获取上下文
    val ssc = new StreamingContext(spark.sparkContext, duration)
    //设置checkpoint
    ssc.checkpoint("C:\Users\80621\Desktop\data\streamData\chk-1")
    //获取数据源
    val lines:DStream[String] = ssc.socketTextStream("user1", 6666)
    //001 mi moblie
    val pairs:DStream[(String, Int)] = lines.map(line => {
      val fields = line.split("\s+")
      if(fields == null || fields.length != 3) {
        ("", -1)
      } else {
        val brand = fields(1)
        val category = fields(2)
        (s"${category}_${brand}", 1)
      }
    }).filter(t => t._2 != -1)
    //实时累加计算
    val usb:DStream[(String, Int)] = pairs.updateStateByKey(updateFunc)
    //将ds数据进行循环
    usb.foreachRDD((rdd, bTime) => {
      if(!rdd.isEmpty()) {//category_brand count
        import spark.implicits._
        //将rdd转换成df
        val df = rdd.map{
          case (cb, count) => {
            val category = cb.substring(0, cb.indexOf("_"))
            val brand = cb.substring(cb.indexOf("_") + 1)
            (category, brand, count)
          }}.toDF("category", "brand", "sales")
        //将df注册成临时表
        df.createOrReplaceTempView("tmp_category_brand_sales")
        //根据category进行分组排名，然后取top3的SQL语句
        val sql =
          """
            |select
            |  t.category,
            |  t.brand,
            |  t.sales,
            |  t.rank
            |from (
            |  select
            |    category,
            |    brand,
            |    sales,
            |    row_number() over(partition by category order by sales desc) rank
            |  from tmp_category_brand_sales
            |) t
            |where t.rank < 4
            |;
                    """.stripMargin
        //执行SQL语句并打印结果
        spark.sql(sql).show()
      }
    })
    //开启streaming
    ssc.start()
    ssc.awaitTermination()
  }

  def updateFunc(seq: Seq[Int], option: Option[Int]): Option[Int] = {
    Option(seq.sum + option.getOrElse(0))
  }
}

显示每个类别的前三名

SparkStreaming学习记录

1.3. Storm V.S. SparkStreaming V.S. Flink

二、SparkStreaming实时处理入门

2.1. 工程创建
在pom.xml文件中导入SparkStreaming的maven依赖

org.apache.spark spark-streaming_2.12 3.1.2

在com package下创建一个新的package named sparkStreaming_example。

三、SparkStreaming与Kafka整合

Java相关栏目本月热门文章

SparkStreaming学习记录

1.3. Storm V.S. SparkStreaming V.S. Flink

二、SparkStreaming实时处理入门

2.1. 工程创建 在pom.xml文件中导入SparkStreaming的maven依赖 org.apache.spark spark-streaming_2.12 3.1.2 在com package下创建一个新的package named sparkStreaming_example。

三、SparkStreaming与Kafka整合

Java相关栏目本月热门文章

2.1. 工程创建
在pom.xml文件中导入SparkStreaming的maven依赖

org.apache.spark spark-streaming_2.12 3.1.2

在com package下创建一个新的package named sparkStreaming_example。