kafka作为一个实时的分布式消息队列，实时的生产和消费消息，
这里我们可以利用SparkStreaming实时计算框架实时地读取kafka中的数据然后进行计算。
在spark1.3版本后，kafkaUtils里面提供了两个创建dstream的方法，
一种为KafkaUtils.createDstream，另一种为KafkaUtils.createDirectStream
KafkaUtils.createDstream方式
构造函数为KafkaUtils.createDstream(ssc, [zk], [consumer group id], [per-topic,partitions] ) 使用了receivers来接收数据，利用的是Kafka高层次的消费者api，对于所有的receivers接收到的数据将会保存在Spark executors中，然后通过Spark Streaming启动job来处理这些数据，默认会丢失，可启用WAL日志，它同步将接受到数据保存到分布式文件系统上比如HDFS。 所以数据在出错的情况下可以恢复出来 。
A、创建一个receiver来对kafka进行定时拉取数据，ssc的rdd分区和kafka的topic分区不是一个概念，故如果增加特定主消费的线程数仅仅是增加一个receiver中消费topic的线程数，并不增加spark的并行处理数据数量。
B、对于不同的group和topic可以使用多个receivers创建不同的DStream
C、如果启用了WAL(spark.streaming.receiver.writeAheadLog.enable=true)
同时需要设置存储级别(默认StorageLevel.MEMORY_AND_DISK_SER_2)，
即KafkaUtils.createStream(….,StorageLevel.MEMORY_AND_DISK_SER)
KafkaUtils.createDstream实战
- 添加kafka的pom依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
<version>2.1.0</version>
</dependency>
-
启动zookeeper集群
zkServer.sh start
-
启动kafka集群
kafka-server-start.sh /export/servers/kafka/config/server.properties
-
创建topic
kafka-topics.sh --create --zookeeper hdp-node-01:2181 --replication-factor 1 --partitions 3 --topic kafka_spark
-
向topic中生产数据
通过shell命令向topic发送消息
kafka-console-producer.sh --broker-list hdp-node-01:9092 --topic kafka_spark
Spark-Streaming整合kafka实战总结: - 编写Spark Streaming应用程序
package cn.bw.kafka
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import scala.collection.immutable
//todo:利用sparkStreaming接受kafka中的数据实现单词计数----采用receivers
object SparkStreamingKafka_Receiver_checkpoint {
def updateFunc(a:Seq[Int], b:Option[Int]) :Option[Int] ={
Some(a.sum+b.getOrElse(0))
}
def main(args: Array[String]): Unit = {
val checkpointPath = "./kafka-receiver"
val ssc = StreamingContext.getOrCreate(checkpointPath, () => {
createFunc(checkpointPath)
})
ssc.start()
ssc.awaitTermination()
}
def createFunc(checkpointPath:String): StreamingContext = {
//todo:1、创建sparkConf
val sparkConf: SparkConf = new SparkConf()
.setAppName("SparkStreamingKafka_Receiver_checkpoint")
.setMaster("local[4]")
//todo:开启wal预写日志
.set("spark.streaming.receiver.writeAheadLog.enable","true")
//todo:2、创建sparkContext
val sc = new SparkContext(sparkConf)
sc.setLogLevel("WARN")
//todo:3、创建StreamingContext
val ssc = new StreamingContext(sc,Seconds(5))
ssc.checkpoint(checkpointPath)
//todo:4、指定zkServer
val zkServer="node1:2181,node2:2181,node3:2181"
//todo:5、指定groupId
val groupId="spark-kafka-receiver01"
//todo:6、指定topics 这个可以利用一个消费者组来消费多个topic,
//(topic_name -> numPartitions) 指定topic消费的线程数
val topics=Map("kafka_spark"->1)
//todo:7、并行运行更多的接收器读取kafak topic中的数据，这里设置3个
val resultDStream: immutable.IndexedSeq[DStream[String]] = (1 to 3).map(x => {
//todo:8、通过使用KafkaUtils的createStream接受kafka topic中的数据，生成DStream
val kafkaDataDStream: DStream[String] = KafkaUtils.createStream(ssc, zkServer, groupId, topics).map(x => x._2)
kafkaDataDStream
}
)
//todo:利用StreamContext将所有的DStream组合在一起
val kafkaDStream: DStream[String] = ssc.union(resultDStream)
//todo:8、获取kafka中topic的内容
//todo:9、切分每一行。每个单词记为1
val wordAndOne: DStream[(String, Int)] = kafkaDStream.flatMap(_.split(" ")).map((_,1))
//todo:10、相同单词出现的次数累加
val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
//todo:打印
result.print()
ssc
}
}
- 运行代码,查看控制台结果数据
Spark-Streaming整合kafka实战总结:
总结:
通过这种方式实现，刚开始的时候系统正常运行，没有发现问题，但是如果系统异常重新启动sparkstreaming程序后，发现程序会重复处理已经处理过的数据，这种基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。
这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次，可能会处理两次。
因为Spark和ZooKeeper之间可能是不同步的。官方现在也已经不推荐这种整合方式，官网相关地址下面我们使用官网推荐的第二种方式kafkaUtils的createDirectStream()方式。