kafka高階資料來源拉取到spark，偏移量自我維護寫入到redis,建立redis連線池。

需要匯入

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
<version>2.2.1</version>
</dependency>
<!--匯入redis的客戶端的java的依賴包，就可以實現和redis的連線。-->
<dependency>
<groupId>redis.clients</

import java.{lang, util}

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

object WCKafkaRedisApp {

  //  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("xx")
      //每秒鐘每個分割槽kafka拉取訊息的速率
      .set("spark.streaming.kafka.maxRatePerPartition", "100")
      // 序列化
      .set("spark.serilizer", "org.apache.spark.serializer.KryoSerializer")
      // 建議開啟rdd的壓縮
      .set("spark.rdd.compress", "true")
    val ssc = new StreamingContext(conf, Seconds(2))

    //啟動一引數設定
    val groupId = "test002"
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hdp01:9092,hdp02:9092,hdp03:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> groupId,
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: lang.Boolean)
    )
    val topics = Array("test")

    //啟動二引數設定
    var formdbOffset: Map[TopicPartition, Long] = JedisOffset(groupId)

    //拉取kafka資料
    val stream: InputDStream[ConsumerRecord[String, String]] = if (formdbOffset.size == 0) {
      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
      )
    } else {
      KafkaUtils.createDirectStream(
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Assign[String, String](formdbOffset.keys, kafkaParams, formdbOffset)
 

      )
    }


    //資料偏移量處理。
    stream.foreachRDD({
      rdd =>
        //獲得偏移量物件陣列
        val offsetRange: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        //邏輯處理
        rdd.flatMap(_.value().split(" ")).map((_, 1)).reduceByKey(_ + _).foreachPartition({
          it =>
            val jedis = RedisUtils.getJedis
            it.foreach({
              v =>
                jedis.hincrBy("wordcount", v._1, v._2.toLong)
            })
            jedis.close()
        })

        //偏移量存入redis
        val jedis: Jedis = RedisUtils.getJedis
        for (or <- offsetRange) {
          jedis.hset(groupId, or.topic + "-" + or.partition, or.untilOffset.toString)
        }
        jedis.close()

    })


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

...

import java.util

import org.apache.kafka.common.TopicPartition

object JedisOffset {


  def apply(groupId: String) = {
    var formdbOffset = Map[TopicPartition, Long]()
    val jedis1 = RedisUtils.getJedis
    val topicPartitionOffset: util.Map[String, String] = jedis1.hgetAll(groupId)
    import scala.collection.JavaConversions._
    val topicPartitionOffsetlist: List[(String, String)] = topicPartitionOffset.toList
    for (topicPL <- topicPartitionOffsetlist) {
      val split: Array[String] = topicPL._1.split("[-]")
      formdbOffset += (new TopicPartition(split(0), split(1).toInt) -> topicPL._2.toLong)
    }
    formdbOffset
  }
}