SparkStreaming整合Kafka--直连方式实现

方式一：

package XXX

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import kafka.utils.{ZKGroupTopicDirs, ZkUtils}
import org.I0Itec.zkclient.ZkClient
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* Create by 。。。
*
*/
object KafkaDirectWordCount {
def main(args: Array[String]): Unit = {

//指定组名
val group = "g001"

//创建SparkConf
val conf = new SparkConf().setAppName("KafkaDirectWordCount").setMaster("local[4]")

//创建SparkStreaming,并设置间隔时间
val ssc = new StreamingContext(conf,Seconds(5))

//指定消费的topic名字
val topic = "wordCount"

//指定Kafka的broker地址（SparkStreaming的Task直接连到Kafka的分区上，用更加底层的API消费，效率更高）
val brokerList = "L3:9092,L4:9092,L5:9092"

//指定zookeeper的地址，后期更新消费的偏移量时使用（以后可以使用Redis,Mysql来记录偏移量）
val zkQuorum = "L1:2181,L2:2181,L3:2181"

//创建stream时使用的topic名字集合，SparkStreaming可同时消费多个topic
val topics: Set[String] = Set(topic)

//创建一个ZKGroupTopicDirs对象，其实是指定往zk中写入数据的目录，用于保存偏移量
val topicDirs = new ZKGroupTopicDirs(group,topic)

//获取zookeeper中的路径"/g001/offsets/wordCount/"
val zkTopicPath = s"${topicDirs.consumerOffsetDir}"

//准备Kafka的参数
val kafkaParams = Map(
"metadata.broker.list" -> brokerList,
"group.id" -> group,
//从头开始读取数据
"auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString
)

//zookeeper的host和ip,创建一个client，用于更新偏移量
//是zookeeper的客户端，可以从zk中读取偏移量数据，并更新偏移量
val zkClient = new ZkClient(zkQuorum)

//查询该路径下是否字节点（默认有字节点为我们自己保存不同 partition 时生成的）
// /g001/offsets/wordCount/0/10001"
// /g001/offsets/wordCount/1/30001"
// /g001/offsets/wordCount/2/10001"
//zkTopicPath  -> /g001/offsets/wordCount/
val children = zkClient.countChildren(zkTopicPath)

var kafkaStream: InputDStream[(String, String)] = null

//如果 zookeeper 中有保存 offset，我们会利用这个 offset 作为 kafkaStream 的起始位置
var fromOffsets: Map[TopicAndPartition, Long] = Map()

//如果保存过 offset
if (children > 0) {
for (i <- 0 until children) {
// /g001/offsets/wordCount/0/10001

// /g001/offsets/wordCount/0
val partitionOffset = zkClient.readData[String](s"$zkTopicPath/$i")
// wordCount/0
val tp = TopicAndPartition(topic, i)
//将不同 partition 对应的 offset 增加到 fromOffsets 中
// wordCount/0 -> 10001
fromOffsets += (tp -> partitionOffset.toLong)
}
//Key: kafka的key   values: "hello tom hello jerry"
//这个会将 kafka 的消息进行 transform，最终 kafka 的数据都会变成 (kafka的key, message) 这样的 tuple
val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())

//通过KafkaUtils创建直连的DStream（fromOffsets参数的作用是:按照前面计算好了的偏移量继续消费数据）
//[String, String, StringDecoder, StringDecoder,     (String, String)]
//  key    value    key的解码方式   value的解码方式
kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
} else {
//如果未保存，根据 kafkaParam 的配置使用最新(largest)或者最旧的（smallest） offset
kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
}

//偏移量的范围
var offsetRanges = Array[OffsetRange]()

//从kafka读取的消息，DStream的Transform方法可以将当前批次的RDD获取出来
//该transform方法计算获取到当前批次RDD,然后将RDD的偏移量取出来，然后在将RDD返回到DStream
val transform: DStream[(String, String)] = kafkaStream.transform { rdd =>
//得到该 rdd 对应 kafka 的消息的 offset
//该RDD是一个KafkaRDD，可以获得偏移量的范围
offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd
}
val messages: DStream[String] = transform.map(_._2)

//依次迭代DStream中的RDD
messages.foreachRDD { rdd =>
//对RDD进行操作，触发Action
rdd.foreachPartition(partition =>
partition.foreach(x => {
println(x)
})
)

for (o <- offsetRanges) {
//  /g001/offsets/wordCount/0
val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
//将该 partition 的 offset 保存到 zookeeper
//  /g001/offsets/wordCount/0/20000
ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)
}
}

ssc.start()
ssc.awaitTermination()

}

}

方式一使用transform获取RDD

方式二：

package XXX

import io.netty.handler.codec.string.StringDecoder
import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.utils.{ZKGroupTopicDirs, ZkUtils}
import org.I0Itec.zkclient.ZkClient
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* Create by 。。。
*
*/
object KafkaDirectWordCountV2 {
def main(args: Array[String]): Unit = {

//指定组名
val group = "g002"

//创建SparkConf
val conf = new SparkConf().setAppName("KafkaDirectWordCountV2").setMaster("local[4]")

//创建SparkStreaming，并设置间隔时间
val ssc = new StreamingContext(conf,Seconds(5))

//指定消费者的topic名字
val topic = "wc"

//指定Kafka的broker地址（SparkStreaming的Task直接连到Kafka的分区上，用更加底层的API消费，效率更高）
val brokerList = "L3:9092,L4:9092,L5:9092"

//指定zk的地址，后期更新消费的偏移量时使用（也可以使用Redis,mysql来记录偏移量）
val zkQuroum = "L1:2181,L2:2181,L3:2181"

//创建stream时使用的topic名字集合，SparkStreaming可同时消费多个topic
val topics: Set[String] = Set(topic)

//创建一个ZKGroupTopicDirs对象，其实是指定往zk中写入数据的目录，用于保存偏移量
val topicDirs = new ZKGroupTopicDirs(group,topic)
//获取zookeeper中的路径"g002/offsets/wc"
val zkTopicPath = s"${topicDirs.consumerOffsetDir}"

//准备Kafka的参数
val kafkaParams = Map(
"metadata.broker.list" -> brokerList,
"group.id" -> group,
//从头开始读取数据
"auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString
)

//zookeeper的host和IP，创建一个client，用于更新偏移量
//是zookeeper的客户端，可以从zk中读取偏移量数据，并更新偏移量
val zkClient = new ZkClient(zkQuroum)

//查询改路径下是否有节点（默认有子节点是为我们保存不同partition时生成的）
// /g002/offsets/wc/0/10001
// /g002/offsets/wc/1/30001
// /g002/offsets/wc/2/10001
// zkTopicPath -> /g002/offsets/wc/
val children = zkClient.countChildren(zkTopicPath)

var kafkaStream:InputDStream[(String,String)] = null

//如果zookeeper中保存有offset，我们会利用这个offset作为kafkaStream的起始位置
var fromOffsets:Map[TopicAndPartition,Long] = Map()

//如果保存过offset
if (children > 0) {
for (i <- 0 until children) {
// /g002/offsets/wc/0/10001

// /g002/offsets/wc/0
val partitionOffset
24000
= zkClient.readData[String](s"$zkTopicPath/$i")
// wc/0
val tp = TopicAndPartition(topic, i)
//将不同 partition 对应的 offset 增加到 fromOffsets 中
// wc/0 -> 10001
fromOffsets += (tp -> partitionOffset.toLong)
}
//Key: kafka的key   values: "hello tom hello jerry"
//这个会将 kafka 的消息进行 transform，最终 kafka 的数据都会变成 (kafka的key, message) 这样的 tuple
val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())

//通过KafkaUtils创建直连的DStream（fromOffsets参数的作用是:按照前面计算好了的偏移量继续消费数据）
//[String, String, StringDecoder, StringDecoder,     (String, String)]
//  key    value    key的解码方式   value的解码方式
kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
} else {
//如果未保存，根据 kafkaParam 的配置使用最新(largest)或者最旧的（smallest） offset
kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
}

//偏移量的范围
var offsetRanges = Array[OffsetRange]()

//直连方式只有在KafkaDStream的RDD中才能获取偏移量，那么就不能到调用DStream的Transformation
//所以只能在kafkaStream调用foreachRDD，获取RDD的偏移量，然后就是对RDD进行操作了
//依次迭代KafkaDStream中的KafkaRDD
kafkaStream.foreachRDD { kafkaRDD =>
//只有KafkaRDD可以强转成HasOffsetRanges，并获取到偏移量
offsetRanges = kafkaRDD.asInstanceOf[HasOffsetRanges].offsetRanges
val lines: RDD[String] = kafkaRDD.map(_._2)

//对RDD进行操作，触发Action
lines.foreachPartition(partition =>
partition.foreach(x => {
println(x)
})
)

for (o <- offsetRanges) {
//  /g002/offsets/wc/0
val zkPath = s"${topicDirs.consumerOffsetDir}/${o.partition}"
//将该 partition 的 offset 保存到 zookeeper
//  /g002/offsets/wc/0/20000
ZkUtils.updatePersistentPath(zkClient, zkPath, o.untilOffset.toString)
}
}

ssc.start()
ssc.awaitTermination()

}

}

方式二使用kafkaStream获取RDD