SparkSteaming中直连与receiver两种方式的区别
SparkSteaming中直连与receiver两种方式的区别
SparkStreaming的Receiver方式和直连方式有什么区别？
Receiver接收固定时间间隔的数据（放在内存中的），使用高级API，自动维护偏移量，达到固定的时间才去进行处理，效率低并且容易丢失数据，灵活性特别差，不好，而且它处理数据的时候，如果某一刻的数据量过大，那么就会造成磁盘溢写的情况，他通过WALS进行磁盘写入。
Receiver实现方式：
代码如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
object KafkaWC02 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("kafkaWC").setMaster("local[2]") //设置线程数
val ssc = new StreamingContext(conf, Seconds(5))
//设置检查点
ssc.checkpoint("D:\\data\\checpoint\\checpoint1")
//接下来编写kafka的配置信息
val zks = "spark01:2181"
//然后是kafka的消费组
val groupId = "gp1"
//Topic的名字 Map的key是Topic名字，第二个参数是线程数
val topics = Map[String, Int]("test02" -> 1)
//创建kafka的输入数据流，来获取kafka中的数据
val data = KafkaUtils.createStream(ssc, zks, groupId, topics)
//获取到的数据是键值对的格式(key,value)
//获取到的数据是 key是偏移量 value是数据
//接下来开始处理数据
val lines = data.flatMap(_._2.split(" "))
val words = lines.map((_, 1))
val res = words.updateStateByKey(updateFunc,new HashPartitioner(ssc.sparkContext.defaultParallelism),true)
res.print()
//val result = words.reduceByKey(_ + _)
//val res = result.updateStateByKey[Int](updateFunc)
//res.print()
//打印输出
//result.print()
//启动程序
ssc.start()
//等待停止
ssc.awaitTermination()
}
//(iterator:Iteratot[(K,Seq[V]),Option[S]]))
//传过来的值是Key Value类型
//第一个参数，是我们从kafka获取到的元素，key ,String类型
//第二个参数，是我们进行单词统计的value值，Int类型
//第三个参数，是我们每次批次提交的中间结果集
val updateFunc=(iter:Iterator[(String,Seq[Int],Option[Int])])=>{
iter.map(t=>{
(t._1,t._2.sum+t._3.getOrElse(0))
})
}
} |
Direct直连方式，
它使用的是底层API实现Offest我们开发人员管理，这样的话，它的灵活性特别好。并且可以保证数据的安全性，而且不用担心数据量过大，因为它有预处理机制，进行提前处理，然后再批次提交任务。
Direct实现方式：
代码如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
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.InputDStream
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.{Duration, StreamingContext}
/** * 重要！！！ Direct直连方式
*/
object KafkaDirectWC {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("Direct").setMaster("local[2]")
val ssc = new StreamingContext(conf,Duration(5000))
//指定组名
val groupId = "gp01"
//指定消费的topic名字
val topic = "tt"
//指定kafka的Broker地址（SparkStreaming的Task直接连接到Kafka分区上，用的是底层API消费）
val brokerList ="spark:9092"
//接下来我们要自己维护offset了，将offset保存到ZK中
val zkQuorum = "spark:2181"
//创建stream时使用的topic名字集合，SparkStreaming可以同时消费多个topic
val topics:Set[String] = Set(topic)
//创建一个ZkGroupTopicDirs对象，其实是指定往Zk中写入数据的目录
// 用于保存偏移量
val TopicDirs = new ZKGroupTopicDirs(groupId,topic)
//获取zookeeper中的路径“/gp01/offset/tt/”
val zkTopicPath = s"${TopicDirs.consumerOffsetDir}"
//准备kafka参数
val kafkas = Map(
"metadata.broker.list"->brokerList,
"group.id"->groupId,
//从头开始读取数据
"auto.offset.reset"->kafka.api.OffsetRequest.SmallestTimeString
)
// zookeeper 的host和ip，创建一个client，用于更新偏移量
// 是zookeeper客户端，可以从zk中读取偏移量数据，并更新偏移量
val zkClient = new ZkClient(zkQuorum)
//"/gp01/offset/tt/0/10001"
//"/gp01/offset/tt/1/20001"
//"/gp01/offset/tt/2/30001"
val clientOffset = zkClient.countChildren(zkTopicPath)
// 创建KafkaStream
var kafkaStream :InputDStream[(String,String)]= null
//如果zookeeper中有保存offset 我们会利用这个offset作为KafkaStream的起始位置
//TopicAndPartition [/gp01/offset/tt/0/ , 8888]
var fromOffsets:Map[TopicAndPartition,Long] = Map()
//如果保存过offset
if(clientOffset > 0){
//clientOffset 的数量其实就是 /gp01/offset/tt的分区数目
for(i<-0 until clientOffset){
// /gp01/offset/tt/ 0/10001
val partitionOffset = zkClient.readData[String](s"$zkTopicPath/${i}")
// tt/0
val tp = TopicAndPartition(topic,i)
//将不同partition 对应得offset增加到fromoffset中
// tt/0 -> 10001
fromOffsets += (tp->partitionOffset.toLong)
}
// key 是kafka的key value 就是kafka数据
// 这个会将kafka的消息进行transform 最终kafka的数据都会变成(kafka的key,message)这样的Tuple
val messageHandler = (mmd:MessageAndMetadata[String,String])=>
(mmd.key(),mmd.message())
// 通过kafkaUtils创建直连的DStream
//[String,String,StringDecoder, StringDecoder,(String,String)]
// key value key解码方式 value的解码方式 接收数据的格式
kafkaStream = KafkaUtils.createDirectStream
[String,String,StringDecoder,
StringDecoder,(String,String)](ssc,kafkas,fromOffsets,messageHandler)
}else{
//如果未保存，根据kafkas的配置使用最新的或者最旧的offset
kafkaStream = KafkaUtils.createDirectStream
[String,String,StringDecoder,StringDecoder](ssc,kafkas,topics)
}
//偏移量范围
var offsetRanges = Array[OffsetRange]()
//从kafka读取的数据，是批次提交的，那么这块注意下，
// 我们每次进行读取数据后，需要更新维护一下偏移量
//那么我们开始进行取值
// val transform = kafkaStream.transform{
// rdd=>
// //得到该RDD对应得kafka消息的offset
// // 然后获取偏移量
// offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// rdd
// }
// val mes = transform.map(_._2)
// 依次迭代DStream中的RDD
kafkaStream.foreachRDD{
//对RDD进行操作 触发Action
kafkardd=>
offsetRanges = kafkardd.asInstanceOf[HasOffsetRanges].offsetRanges
//下面 你就可以怎么写都行了，为所欲为
val maps = kafkardd.map(_._2)
maps.foreach(println)
for(o<-offsetRanges){
// /gp01/offset/tt/ 0
val zkpath = s"${TopicDirs.consumerOffsetDir}/${o.partition}"
//将该partition的offset保存到zookeeper中
// /gp01/offset/tt/ 0/88889
ZkUtils.updatePersistentPath(zkClient,zkpath,o.untilOffset.toString)
}
}
// 启动
ssc.start()
ssc.awaitTermination()
}
} |