大数据学习day32-----spark12-----1. sparkstreaming（1.1简介，1.2 sparkstreaming入门程序（统计单词个数，updateStageByKey的用法，1.3 SparkStreaming整合Kafka，1.4 SparkStreaming获取KafkaRDD的偏移量，并将偏移量写入kafka中）

1. Spark Streaming

1.1 简介（来源：spark官网介绍）

    Spark Streaming是Spark Core API的扩展，其是支持可伸缩、高吞吐量、容错的实时数据流处理。Spark Streaming的数据源可以为kafka，Flume，Kinesis或者是TCP socket，并且这些数据可以使用复杂的算法来处理，这些算法用高级函数表示，如map、reduce、join和window。最后被处理的数据可以被push到文件存储系统，数据库，live dashboards。实际上，我们也可以使用spark的machine learning和graph processing算法处理数据流

 实际上，Spark Streaming是如下工作的。Spark Streaming 接收实时数据并将这些数据划分成多批小批次数据，然后由spark引擎处理，以批量生成最终的结果流。

 Spark流提供了一种高级抽象，称为离散流或DStream，它表示连续的数据流。DStreams可以从Kafka、Flume和Kinesis等源的输入数据流创建，也可以通过对其他DStreams应用高级操作创建。实际上，DStream内部为RDDs序列。

 

补充：sparkstreaming是一个高容错，可扩展，编程api丰富的分布式实时流计算框架，其是微批次、准实时的

高容错：当某个task所在的机器挂掉之后，其会被调度到其他的机器上去运行。而task是自己维护读取数据偏移量的，若task不运行完，其实不会记录偏移量的（事务）。所以是高容错的

高吞吐：spark读取完数据可以先放内存中（装满放磁盘），所以其可以将大量的数据存到内存中去

可扩展：当数据量增加时，可以增加rdd的数量来加快运算

编程api丰富：这是和storm比较，若是和flink比较则没有这个优点

注意：不管有没有数据，只要到了自己设置批次的时间（new StreamingContext时设置），driver端就会调度一批任务去executor中去。每个批次的任务处理的数据都是这段时间读取的数据

1.2 Spark Streaming入门程序

需求：利用Spark Streaming实时统计单词的个数

linux上安装nc：

yum -y install nc 

在安装有nc的节点上开通一个socket服务

nc -lk 8888

SparkStreamingWordCount

object SparkStreamingWordCount {
  def main(args: Array[String]): Unit = {
    // 创建sparkcontext
    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc= new SparkContext(conf)
    // 设置日志的级别
    sc.setLogLevel("WARN")
    // StreamingContext是对SparkContext包装和增强，后面传入的时间代表隔多久时间产生一个批次
    val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))
    // 获取Dstream，ReceiverInputDStream为Dstream的孙子类
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("feng05",8888)
    // 对Dstream进行操作
    val reduced: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
    //Action
    reduced.print()
    //开启spark streaming
    ssc.start()
    //让程序挂起，一直运行
    ssc.awaitTermination()
  }
}

此时运行该程序，然后在socket服务中心输入一些单词，如下

 在程序中会做出单词统计（此种形式只会统计当前数据，不会累加历史的数据），如下（对应最后一行）

 

 现在需要累加历史批次的数据，代码可写成如下

UpdateStageWordCount 

object UpdateStageWordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    //创建StreamingContext，并指定批次生成的时间
    val ssc = new StreamingContext(conf, Milliseconds(5000))
    //设置日志级别
    ssc.sparkContext.setLogLevel("WARN")
    //对数据进行处理，累加历史批次的数据
    //累加历史数据就要将中间结果保存起来【要求保存到靠谱的存储系统（安全），以后任务失败可以恢复State数据】
    //设置checkpoint目录，保存历史状态
    ssc.checkpoint("file:///javafile/spark/ck")
    //创建DStream
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("feng05", 8889)
    //Transformation 开始
    val words: DStream[String] = lines.flatMap(_.split(" "))
    val wordAndOne: DStream[(String, Int)] = words.map((_, 1))
    //updateFunc: (Seq[V], Option[S]) => Option[S]
    //第一个参数：当前批次计算的结果
    //第二个参数：初始值或中间累加结果
    val updateFunc: (Seq[Int], Option[Int]) => Some[Int] = (s: Seq[Int], o: Option[Int]) => {
      //将当前批次累加的结果和初始值或中间累加结果进行累加
      Some(s.sum + o.getOrElse(0))
    }
    val reduced: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
    //Transformation 结束
    //触发Action
    reduced.print()
    //开启Streaming程序
    ssc.start()
    //挂起一直运行
    ssc.awaitTermination()
  }
}

结果

 此处注意的两个点：（1）将中间结果checkpoint保存起来（2）updateStateByKey的用法

updateStateByKey()的用法

源码中有7中重载的方法，此处讲传入参数仅为一个函数的重载方法（最简单的形式）

 /**
   * Return a new "state" DStream where the state for each key is updated by applying
   * the given function on the previous state of the key and the new values of each key.
   * In every batch the updateFunc will be called for each state even if there are no new values.
   * Hash partitioning is used to generate the RDDs with Spark's default number of partitions.
   * @param updateFunc State update function. If `this` function returns None, then
   *                   corresponding state key-value pair will be eliminated.
   * @tparam S State type
   */
  def updateStateByKey[S: ClassTag](
      updateFunc: (Seq[V], Option[S]) => Option[S]
    ): DStream[(K, S)] = ssc.withScope {
    updateStateByKey(updateFunc, defaultPartitioner())   // 没定义分区器时，使用默认的分区器
  }

中文翻译：

　　返回一个新的“状态”DStream，其中通过对键的前一个状态和每个键的新值应用给定的函数来更新每个键的状态。在每个批处理中，即使没有新值，也会为每个状态调用updateFunc。散列分区用于生成具有Spark默认分区数的RDDs。

updateFunc函数中：

　　第一个参数：当前批次每个key新增值的集合(key为调用updateStateByKey的DStream中的key)，如当新输入的单词为：hadoop hadoop spark flink，则此集合为[2,1,1]

　　第二个参数：当前键保存的前一个状态（本例中即为初始值或中间累加结果）

补充：（第二遍的理解）

//第一个参数：表示当前key对应的所有值
//第二个参数：Option[S] 是当前key的历史状态，返回的是新的封装的数据。

 这块源码还有些看不懂（以后回头看）

 foreachRdd：

　　其实一个普通的方法，既不是active算子也不是转化算子，用来取dstream中的rdd

1.3 SparkStreaming整合Kafka

SparkStreaming跟kafka进行整合

• 导入跟kafka整合的依赖
• 跟kafka整合，创建直连的DStream【即使用createDirectStream创建，使用底层的消费API，效率更高】

　　 　　消费者直接连接到kafka的leader分区进行消费

　　　　 直连的方式，RDD的分区数量和kafka的分区数量是一一对应的【数目一样】

官网上对这个新接口（createDirectStream）的介绍很多，大致就是不与zookeeper交互，直接去kafka中读取数据，自己维护offset，于是速度比KafkaUtils.createStream要快上很多。但有利就有弊：无法进行offset的监控。

两篇相关博客：

　　http://blog.selfup.cn/1665.html

　　https://www.cnblogs.com/hd-zg/p/6188287.html

（1）整合　　

package com._51doit

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

object KafkaStreamingWordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    //创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Milliseconds(5000))
    // 设置日志级别
    ssc.sparkContext.setLogLevel("WARN")
    val topics = Array("wordcount")

    //SparkSteaming跟kafka整合的参数
    //kafka的消费者默认的参数就是每5秒钟自动提交偏移量到Kafka特殊的topic中: __consumer_offsets
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "feng05:9092,feng06:9092,feng07:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "g005",
      "auto.offset.reset" -> "earliest"//如果没有记录偏移量，第一次从最开始读，有偏移量，接着偏移量读
      //, "enable.auto.commit" -> (false: java.lang.Boolean) //消费者不自动提交偏移量
    )
    // 跟kafka进行整合，需要引入跟kafka整合的依赖
    //createDirectStream更加高效，使用的是kafka底层的消费API，消费者直接连接到Kafka的Leader分区进行消费
    //直连方式，RDD的分区数量和Kafka的分区数量是一一对应的【数目一样】
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent, //调度task到Kafka所在的节点
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //指定订阅Topic的规则
    )
//    kafkaDStream.print()
//    val lines: DStream[String] = kafkaDStream.map(r=>r.key())
    val lines: DStream[String] = kafkaDStream.map(r => r.value())
    lines.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

（2）将统计到的数据统计写入redis

package com._51doit

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import redis.clients.jedis.Jedis

object KafkaStreamingWordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    //创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Milliseconds(5000))
    // 设置日志级别
    ssc.sparkContext.setLogLevel("WARN")
    val topics = Array("wordcount")

    //SparkSteaming跟kafka整合的参数
    //kafka的消费者默认的参数就是每5秒钟自动提交偏移量到Kafka特殊的topic中: __consumer_offsets
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "feng05:9092,feng06:9092,feng07:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "g01",
      "auto.offset.reset" -> "earliest"//如果没有记录偏移量，第一次从最开始读，有偏移量，接着偏移量读
      //, "enable.auto.commit" -> (false: java.lang.Boolean) //消费者不自动提交偏移量
    )
    // 跟kafka进行整合，需要引入跟kafka整合的依赖
    //createDirectStream更加高效，使用的是kafka底层的消费API，消费者直接连接到Kafka的Leader分区进行消费
    //直连方式，RDD的分区数量和Kafka的分区数量是一一对应的【数目一样】
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent, //调度task到Kafka所在的节点
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //指定订阅Topic的规则
    )
//    kafkaDStream.print()
//    val lines: DStream[String] = kafkaDStream.map(r=>r.key())
    val lines: DStream[String] = kafkaDStream.map(r => r.value())
    val words: DStream[String] = lines.flatMap(_.split(" "))
    val WordAndOne: DStream[(String, Int)] = words.map((_,1))
    // 计算当前批次
    val reduced: DStream[(String, Int)] = WordAndOne.reduceByKey(_+_)
    // 将当前批次数据和历史批次数据进行累加
    reduced.foreachRDD(rdd => {
      // 将数据收集到driver端
      val res: Array[(String, Int)] = rdd.collect()
      //将数据写入到Redis
      //创建Jedis
      val jedis: Jedis = new Jedis("feng05", 6379)
      jedis.auth("feng")
      jedis.select(2)
      res.foreach(t => {
        jedis.hincrBy("Streaming_wordcount", t._1, t._2)
      })
      jedis.close()
    })
//    lines.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

 1.4 SparkStreaming获取KafkaRDD的偏移量，并将偏移量写入kafka中

JedisConnectionPool（redis连接池代码）

package utils

import redis.clients.jedis.{Jedis, JedisPool, JedisPoolConfig}

object JedisConnectionPool {
  private val config: JedisPoolConfig = new JedisPoolConfig()
  config.setMaxTotal(20)  //设置最大的连接数
  config.setMaxIdle(10)   // 设置最大空闲连接数
  //timeout等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。
  private val pool: JedisPool = new JedisPool(config, "feng05", 6379,10000,"feng")
  def getConnection: Jedis ={
    pool.getResource
  }
}

业务代码

package com._51doit

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import redis.clients.jedis.Jedis
import utils.JedisConnectionPool

object KafkaStreamingWordCountManageOffset {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    //创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Milliseconds(5000))
    // 设置日志级别
    ssc.sparkContext.setLogLevel("WARN")
    val topics = Array("wordcount")

    //SparkSteaming跟kafka整合的参数
    //kafka的消费者默认的参数就是每5秒钟自动提交偏移量到Kafka特殊的topic中: __consumer_offsets
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "feng05:9092,feng06:9092,feng07:9092",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "group.id" -> "g03",
      "auto.offset.reset" -> "earliest"//如果没有记录偏移量，第一次从最开始读，有偏移量，接着偏移量读
      //, "enable.auto.commit" -> (false: java.lang.Boolean) //消费者不自动提交偏移量
    )
    // 跟kafka进行整合，需要引入跟kafka整合的依赖
    //createDirectStream更加高效，使用的是kafka底层的消费API，消费者直接连接到Kafka的Leader分区进行消费
    //直连方式，RDD的分区数量和Kafka的分区数量是一一对应的【数目一样】
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent, //调度task到Kafka所在的节点
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) //指定订阅Topic的规则
    )
    // 调用完createDirectStream后，直接用kafkaDStream调用foreeachRDD，，只有KafkaRDD中有偏移量
    kafkaDStream.foreachRDD(rdd => {
//      println(rdd.collect().toBuffer)
      // 计算当前批次的数据
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      val lines: RDD[String] = rdd.map(r => r.value())
      val words: RDD[String] = lines.flatMap(_.split(" "))
      val WordAndOne: RDD[(String, Int)] = words.map((_,1))
      val reduced: RDD[(String, Int)] = WordAndOne.reduceByKey(_+_)
      // 触发action
      reduced.foreachPartition(it => {
        //在Executor端获取redis连接
        val jedis: Jedis = JedisConnectionPool.getConnection
        jedis.select(1)
        // 将分区中的结果写入redis
        it.foreach(t => {
          jedis.hincrBy("Streaming_wordcount", t._1, t._2)
        })
        // 将连接还回redis连接池
        jedis.close()
      })
      //再更新这个批次每个分区的偏移量
      //异步提交偏移量，将偏移量写入到Kafka特殊的topic中了
      kafkaDStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
    })
    ssc.start()
    ssc.awaitTermination()
  }
}

注意：只有kafkardd中有偏移量，若想获取偏移量，则需要通过此rdd获取

 　　

　　

asdas