064 SparkStream与kafka的集成，主要是编程

　　这里面包含了如何在kafka+sparkStreaming集成后的开发，也包含了一部分的优化。

一：说明

1.官网

　　指导网址：http://spark.apache.org/docs/1.6.1/streaming-kafka-integration.html

　　

2.SparkStream+kafka

　　Use Receiver

　　　　内部使用kafka的high lenel consumer API

　　　　consumer offset 只能保持到zk/kafka中，只能通过配置进行offset的相关操作

　　Direct

　　　　内部使用的是kafka的simple consumer api

　　　　自定义对kafka的offset偏移量进行控制操作

　　　　集成依赖pom配置：

　　　　　　

二：单Receiver的程序

1.先启动服务

　　在这里需要启动kafka的生产者

　　

2.程序

package com.stream.it

import kafka.serializer.StringDecoder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object KafkaWordcount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf()
        .setAppName("spark-streaming-wordcount")
          .setMaster("local[*]")
    val sc=SparkContext.getOrCreate(conf)
    val ssc=new StreamingContext(sc,Seconds(15))

    /*
    def createStream[K: ClassTag, V: ClassTag, U <: Decoder[_]: ClassTag, T <: Decoder[_]: ClassTag](
      ssc: StreamingContext,
      kafkaParams: Map[String, String],
      topics: Map[String, Int],
      storageLevel: StorageLevel
    ): ReceiverInputDStream[(K, V)]
    */
    val kafkaParams=Map("group.id"->"stream-sparking-0",
          "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",
          "auto.offset.reset"->"smallest"
    )
    val topics=Map("beifeng"->1)
    val dStream=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
      ssc,             //给定sparkStreaming的上下文
      kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接
      topics,          //给定读取对应的topic的名称以及读取数据的线程数量
      StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别
    ).map(_._2)


    val resultWordcount=dStream
      .filter(line=>line.nonEmpty)
        .flatMap(line=>line.split(" ").map((_,1)))
        .reduceByKey(_+_)
    resultWordcount.foreachRDD(rdd=>{
      rdd.foreachPartition(iter=>iter.foreach(println))
    })

    //启动
    ssc.start()
    //等到
    ssc.awaitTermination()
  }
}

3.效果

　　在kafka producer输入内容，将会在控制台上进行展示

三：多Receiver

1.说明

　　当当个reveiver接收的数据被限制的时候，可以使用多个receiver

2.程序

package com.stream.it

import kafka.serializer.StringDecoder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object MulReceiverKafkaWordcount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf()
        .setAppName("spark-streaming-wordcount2")
          .setMaster("local[*]")
    val sc=SparkContext.getOrCreate(conf)
    val ssc=new StreamingContext(sc,Seconds(15))

    /*
    def createStream[K: ClassTag, V: ClassTag, U <: Decoder[_]: ClassTag, T <: Decoder[_]: ClassTag](
      ssc: StreamingContext,
      kafkaParams: Map[String, String],
      topics: Map[String, Int],
      storageLevel: StorageLevel
    ): ReceiverInputDStream[(K, V)]
    */
    val kafkaParams=Map("group.id"->"stream-sparking-0",
          "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",
          "auto.offset.reset"->"smallest"
    )
    val topics=Map("beifeng"->4)
    val dStream1=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
      ssc,             //给定sparkStreaming的上下文
      kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接
      topics,          //给定读取对应的topic的名称以及读取数据的线程数量
      StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别
    ).map(_._2)

    val dStream2=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
      ssc,             //给定sparkStreaming的上下文
      kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接
      topics,          //给定读取对应的topic的名称以及读取数据的线程数量
      StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别
    ).map(_._2)

    //合并dstream
    val dStream=dStream1.union(dStream2)


    val resultWordcount=dStream
      .filter(line=>line.nonEmpty)
        .flatMap(line=>line.split(" ").map((_,1)))
        .reduceByKey(_+_)
    resultWordcount.foreachRDD(rdd=>{
      rdd.foreachPartition(iter=>iter.foreach(println))
    })

    //启动
    ssc.start()
    //等到
    ssc.awaitTermination()
  }
}

3.效果 

　　一条数据是一个event

　　

 　　这里有两个receiver。

　　

四：Direct

1.说明

　　直接读取，不存在receiver

　　不足，kafkaParams指定连接kafka的参数，内部使用的是kafka的SimpleConsumerAPI，所以，offset只能从头或者从尾开始读取，不能设置。

　　topics：topic的名称

2.程序

package com.stream.it

import kafka.serializer.StringDecoder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object DirectKafkaWordcount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf()
        .setAppName("spark-streaming-wordcount")
          .setMaster("local[*]")
    val sc=SparkContext.getOrCreate(conf)
    val ssc=new StreamingContext(sc,Seconds(15))

    val kafkaParams=Map(
          "metadata.broker.list"->"linux-hadoop01.ibeifeng.com:9092,linux-hadoop01.ibeifeng.com:9093,linux-hadoop01.ibeifeng.com:9094",
          "auto.offset.reset"->"smallest"
    )
    val topics=Set("beifeng")
    val dStream=KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](
      ssc,
      kafkaParams,
      topics).map(_._2)
    
    val resultWordcount=dStream
      .filter(line=>line.nonEmpty)
        .flatMap(line=>line.split(" ").map((_,1)))
        .reduceByKey(_+_)
    resultWordcount.foreachRDD(rdd=>{
      rdd.foreachPartition(iter=>iter.foreach(println))
    })

    //启动
    ssc.start()
    //等到
    ssc.awaitTermination()
  }
}

3.效果

　　没有receiver。

　　

　     　

五：Direct实现是累加器管理offset偏移量

1.程序

　 kafkaParams 中只有这个参数下才能生效。

　　数据先进行保存或者打印，然后更新accumulable中的offset，然后下一批的dstream进行更新offset。

　　累加器需要在外面进行定义。

package com.stream.it

import scala.collection.mutable
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{Accumulable, AccumulableParam, SparkConf, SparkContext}

object AccumubaleKafkaWordcount {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf()
        .setAppName("spark-streaming-wordcount")
          .setMaster("local[*]")
    val sc=SparkContext.getOrCreate(conf)
    val ssc=new StreamingContext(sc,Seconds(15))
    val accumu = DroppedAccumulable.getInstance(sc)

    val kafkaParams = Map(
      "metadata.broker.list" -> "linux-hadoop01.ibeifeng.com:9092,linux-hadoop01.ibeifeng.com:9093,linux-hadoop01.ibeifeng.com:9094,linux-hadoop01.ibeifeng.com:9095"
    )

    // TODO: 从某一个存储offset的地方读取offset偏移量数据, redishbase其他地方.....
    val fromOffsets = Map(
      TopicAndPartition("beifeng", 0) -> -1L, // 如果这里给定的偏移量是异常的，会直接从kafka中读取偏移量数据(largest)
      TopicAndPartition("beifeng", 1) -> 0L,
      TopicAndPartition("beifeng", 2) -> 0L,
      TopicAndPartition("beifeng", 3) -> 0L
    )


    val dstream = KafkaUtils.createDirectStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder, String](
      ssc, // 上下文
      kafkaParams, // kafka连接
      fromOffsets,
      (message: MessageAndMetadata[String, String]) => {
        // 这一块在Executor上被执行
        // 更新偏移量offset
        val topic = message.topic
        val paritionID = message.partition
        val offset = message.offset
        accumu += (topic, paritionID) -> offset
        // 返回value的数据
        message.message()
      }
    )

    val resultWordCount = dstream
      .filter(line => line.nonEmpty)
      .flatMap(line => line.split(" ").map((_, 1)))
      .reduceByKey(_ + _)


    resultWordCount.foreachRDD(rdd => {
      // 在driver上执行
      try {
        rdd.foreachPartition(iter => {
          // 代码在executor上执行
          // TODO: 这里进行具体的数据保存操作
          iter.foreach(println)
        })

        // TODO: 在这里更新offset, 将数据写入到redishbase其他地方.....
        accumu.value.foreach(println)
      } catch {
        case e: Exception => // nothings
      }
    })



    //启动
    ssc.start()
    //等到
    ssc.awaitTermination()
  }
}
object DroppedAccumulable {
  private var instance: Accumulable[mutable.Map[(String, Int), Long], ((String, Int), Long)] = null

  def getInstance(sc: SparkContext): Accumulable[mutable.Map[(String, Int), Long], ((String, Int), Long)] = {
    if (instance == null) {
      synchronized {
        if (instance == null) instance = sc.accumulable(mutable.Map[(String, Int), Long]())(param = new AccumulableParam[mutable.Map[(String, Int), Long], ((String, Int), Long)]() {
          /**
            * 将t添加到r中
            *
            * @param r
            * @param t
            * @return
            */
          override def addAccumulator(r: mutable.Map[(String, Int), Long], t: ((String, Int), Long)): mutable.Map[(String, Int), Long] = {
            val oldOffset = r.getOrElse(t._1, t._2)
            if (t._2 >= oldOffset) r += t
            else r
          }

          override def addInPlace(r1: mutable.Map[(String, Int), Long], r2: mutable.Map[(String, Int), Long]): mutable.Map[(String, Int), Long] = {
            r2.foldLeft(r1)((r, t) => {
              val oldOffset = r.getOrElse(t._1, t._2)
              if (t._2 >= oldOffset) r += t
              else r
            })
          }

          override def zero(initialValue: mutable.Map[(String, Int), Long]): mutable.Map[(String, Int), Long] = mutable.Map.empty[(String, Int), Long]
        })
      }
    }

    // 返回结果
    instance
  }
}

2.效果

　　可以将以前的信息打出来。