超详细！一文详解 SparkStreaming 如何整合 Kafka！附代码可实践

来源 | Alice菌

责编 | Carol

封图 | CSDN 下载于视觉中国

出品 | CSDN（ID:CSDNnews）

相信很多小伙伴已经接触过 SparkStreaming 了，理论就不讲太多了，今天的内容主要是为大家带来的是 SparkStreaming 整合 Kafka 的教程。

文中含代码，感兴趣的朋友可以复制动手试试！

Kafka回顾

正式开始之前，先让我们来对Kafka回顾一波。

核心概念图解

Broker： 安装Kafka服务的机器就是一个broker

Producer： 消息的生产者，负责将数据写入到broker中(push) Consumer： 消息的消费者，负责从kafka中拉取数据(pull)，老版本的消费者需要依赖zk，新版本的不需要 Topic: 主题，相当于是数据的一个分类，不同topic存放不同业务的数据 –主题:区分业务 Replication： 副本，数据保存多少份(保证数据不丢失) –副本:数据安全 Partition： 分区，是一个物理的分区,一个分区就是一个文件,一个Topic可以有1~n个分区,每个分区都有自己的副本 –分区:并发读写 Consumer Group： 消费者组，一个topic可以有多个消费者/组同时消费，多个消费者如果在一个消费者组中，那么他们不能重复消费数据 –消费者组:提高消费者消费速度、方便统一管理 注意[1]： 一个Topic可以被多个消费者或者组订阅，一个消费者/组也可以订阅多个主题 注意[2]： 读数据只能从Leader读，写数据也只能往Leader写，Follower会从Leader那里同步数据过来做副本！！！

常用命令

启动kafka

/ export /servers/kafka/bin/kafka-server-start.sh -daemon  
 
/export/servers/kafka/ config /server.properties

停止kafka

/export/servers/kafka/bin/kafka-server-stop.sh

查看topic信息

/export/servers/kafka/bin/kafka-topics.sh --list -- ZooKeeper  node01:2181

创建topic

/export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 3 --partitions 3 --topic test

查看某个topic信息

/export/servers/kafka/bin/kafka-topics.sh --describe --zookeeper node01:2181 --topic test

删除topic

/export/servers/kafka/bin/kafka-topics.sh --zookeeper node01:2181 --delete --topic test

启动生产者–控制台的生产者一般用于测试

/export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092 --topic spark_kafka

启动消费者–控制台的消费者一般用于测试

/export/servers/kafka/bin/kafka-console-consumer.sh --zookeeper node01:2181 --topic  Spark _kafka--from-beginning

消费者连接到borker的地址

 /export/servers/kafka/bin/kafka-console-consumer.sh --bootstrap-server node01:9092,node02:9092,node03:9092 --topic spark_kafka --from-beginning  
 
   
整合kafka两种模式说明 
这同时也是一个面试题的热点。 
开发中我们经常会利用SparkStreaming实时地读取kafka中的数据然后进行处理，在spark1.3版本后，kafkaUtils里面提供了两种创建DStream的方法: 
 1、Receiver接收方式：  
 KafkaUtils.createDstream （开发中不用，了解即可，但是面试可能会问）。 
Receiver作为常驻的Task运行在Executor等待数据，但是一个Receiver效率低，需要开启多个，再手动合并数据(union)，再进行处理，很麻烦 
Receiver哪台机器挂了，可能会丢失数据，所以需要开启WAL(预写日志)保证数据安全，那么效率又会降低! 
Receiver方式是通过 zookeeper 来连接kafka队列，调用Kafka高阶API，offset存储在zookeeper，由Receiver维护。 
spark在消费的时候为了保证数据不丢也会在 checkpoint 中存一份offset，可能会出现数据不一致 
所以不管从何种角度来说，Receiver模式都不适合在开发中使用了,已经淘汰了 
 2、 Direct 直连方式  
 KafkaUtils.createDirectStream （开发中使用，要求掌握） 
Direct方式是直接连接kafka分区来获取数据，从每个分区直接读取数据大大提高了并行能力 
Direct方式调用Kafka低阶API(底层API)，offset自己存储和维护，默认由Spark维护在checkpoint中，消除了与zk不一致的情况 
当然也可以自己手动维护，把offset存在mysql、redis中 
所以基于Direct模式可以在开发中使用，且借助Direct模式的特点+手动操作可以保证数据的Exactly once 精准一次 
 总结：  
 Receiver接收方式  
多个Receiver接受数据效率高，但有丢失数据的风险 
开启日志（WAL）可防止数据丢失，但写两遍数据效率低。 
Zookeeper维护offset有重复消费数据可能。 
使用高层次的API 
 Direct直连方式  
不使用Receiver，直接到kafka分区中读取数据 
不使用日志（WAL）机制 
Spark自己维护offset 
使用低层次的API 
扩展：关于消息语义 
   
 注意：  
开发中SparkStreaming和kafka集成有两个版本：0.8及0.10+ 
0.8版本有Receiver和Direct模式（但是0.8版本生产环境问题较多，在Spark2.3之后不支持0.8版本了）。 
0.10以后只保留了direct模式(Reveiver模式不适合生产环境)，并且0.10版本API有变化（更加强大） 
   
 结论：  
我们学习和开发都直接使用0.10版本中的direct模式，但是关于Receiver和Direct的区别面试的时候要能够答得上来 
   
 spark-streaming-kafka-0-8(了解)  
 1.Receiver  
KafkaUtils.createDstream使用了receivers来接收数据，利用的是Kafka高层次的消费者api，偏移量由Receiver维护在zk中，对于所有的receivers接收到的数据将会保存在Spark executors中，然后通过Spark Streaming启动job来处理这些数据，默认会丢失，可启用WAL日志，它同步将接受到数据保存到分布式文件系统上比如HDFS。保证数据在出错的情况下可以恢复出来。尽管这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是启用了WAL效率会较低，且无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。 
（官方现在已经不推荐这种整合方式。） 
   
 准备工作  
1）启动zookeeper集群 
 zkServer.sh start 
 
2）启动kafka集群 
kafka-server-start.sh /export/servers/kafka/config/server.properties 
 
3.创建topic 
kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 1 --partitions 3 --topic spark_kafka 
 
4.通过shell命令向topic发送消息 
kafka-console-producer.sh --broker-list node01:9092 --topic spark_kafka 
 
5.添加kafka的pom依赖 
<dependency> 
<groupId>org.apache.spark</groupId> 
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId> 
<version>2.2.0</version> 
</dependency> 
 API  
通过receiver接收器获取kafka中topic数据，可以并行运行更多的接收器读取kafak topic中的数据，这里为3个 
 val receiverDStream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => { 
val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics) 
stream 
}) 
 
如果启用了WAL(spark.streaming.receiver.writeAheadLog.enable=true)可以设置存储级别(默认StorageLevel.MEMORY_AND_DISK_SER_2) 
代码演示 
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} 
import org.apache.spark.streaming.kafka.KafkaUtils 
import org.apache.spark.streaming.{Seconds, StreamingContext} 
import org.apache.spark.{SparkConf, SparkContext} 
 
import scala.collection.immutable 
 
object SparkKafka { 
def main(args: Array[String]): Unit = { 
//1.创建StreamingContext 
val config: SparkConf = 
new SparkConf.setAppName("SparkStream").setMaster("local[*]") 
.set("spark.streaming.receiver.writeAheadLog.enable", "true") 
//开启WAL预写日志，保证数据源端可靠性 
val sc = new SparkContext(config) 
sc.setLogLevel("WARN") 
val ssc = new StreamingContext(sc,Seconds(5)) 
ssc.checkpoint("./kafka") 
//============================================== 
//2.准备配置参数 
val zkQuorum = "node01:2181,node02:2181,node03:2181" 
val groupId = "spark" 
val topics = Map("spark_kafka" -> 2)//2表示每一个topic对应分区都采用2个线程去消费, 
//ssc的rdd分区和kafka的topic分区不一样，增加消费线程数，并不增加spark的并行处理数据数量 
//3.通过receiver接收器获取kafka中topic数据，可以并行运行更多的接收器读取kafak topic中的数据，这里为3个 
val receiverDStream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => { 
val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics) 
stream 
}) 
//4.使用union方法，将所有receiver接受器产生的Dstream进行合并 
val allDStream: DStream[(String, String)] = ssc.union(receiverDStream) 
//5.获取topic的数据(String, String) 第1个String表示topic的名称，第2个String表示topic的数据 
val data: DStream[String] = allDStream.map(_._2) 
//============================================== 
//6.WordCount 
val words: DStream[String] = data.flatMap(_.split(" ")) 
val wordAndOne: DStream[(String, Int)] = words.map((_, 1)) 
val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _) 
result.print 
ssc.start 
ssc.awaitTermination 
} 
} 
 
 2.Direct  
Direct方式会定期地从kafka的topic下对应的partition中查询最新的偏移量，再根据偏移量范围在每个batch里面处理数据，Spark通过调用kafka简单的消费者API读取一定范围的数据。 
   
Direct的缺点是无法使用基于zookeeper的kafka监控工具 
Direct相比基于Receiver方式有几个优点： 
 简化并行 不需要创建多个kafka输入流，然后union它们，sparkStreaming将会创建和kafka分区数一样的rdd的分区数，而且会从kafka中并行读取数据，spark中RDD的分区数和kafka中的分区数据是一一对应的关系。 
 高效 Receiver实现数据的零丢失是将数据预先保存在WAL中，会复制一遍数据，会导致数据被拷贝两次，第一次是被kafka复制，另一次是写到WAL中。而Direct不使用WAL消除了这个问题。 
 恰好一次语义(Exactly-once-semantics) Receiver读取kafka数据是通过kafka高层次api把偏移量写入zookeeper中，虽然这种方法可以通过数据保存在WAL中保证数据不丢失，但是可能会因为sparkStreaming和ZK中保存的偏移量不一致而导致数据被消费了多次。 
Direct的Exactly-once-semantics(EOS)通过实现kafka低层次api，偏移量仅仅被ssc保存在checkpoint中，消除了zk和ssc偏移量不一致的问题。 
 API  
KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) 
 
代码演示 
import kafka.serializer.StringDecoder 
import org.apache.spark.streaming.dstream.{DStream, InputDStream} 
import org.apache.spark.streaming.kafka.KafkaUtils 
import org.apache.spark.streaming.{Seconds, StreamingContext} 
import org.apache.spark.{SparkConf, SparkContext} 
 
 
object SparkKafka2 { 
def main(args: Array[String]): Unit = { 
//1.创建StreamingContext 
val config: SparkConf = 
new SparkConf.setAppName("SparkStream").setMaster("local[*]") 
val sc = new SparkContext(config) 
sc.setLogLevel("WARN") 
val ssc = new StreamingContext(sc,Seconds(5)) 
ssc.checkpoint("./kafka") 
//============================================== 
//2.准备配置参数 
val kafkaParams = Map("metadata.broker.list" -> "node01:9092,node02:9092,node03:9092", "group.id" -> "spark") 
val topics = Set("spark_kafka") 
val allDStream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) 
//3.获取topic的数据 
val data: DStream[String] = allDStream.map(_._2) 
//============================================== 
//WordCount 
val words: DStream[String] = data.flatMap(_.split(" ")) 
val wordAndOne: DStream[(String, Int)] = words.map((_, 1)) 
val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _) 
result.print 
ssc.start 
ssc.awaitTermination 
} 
} 
 
   
spark-streaming-kafka-0-10 
 说明  
spark-streaming-kafka-0-10版本中，API有一定的变化，操作更加灵活，开发中使用 
 pom.xml  
<!--<dependency> 
<groupId>org.apache.spark</groupId> 
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId> 
<version>${spark.version}</version> 
</dependency>--> 
<dependency> 
<groupId>org.apache.spark</groupId> 
<artifactId>spark-streaming-kafka-0-10_2.11</artifactId> 
<version>${spark.version}</version> 
</dependency> 
API： 
   
创建topic 
/export/servers/kafka/bin/kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 3 --partitions 3 --topic spark_kafka 
 
启动生产者 
/export/servers/kafka/bin/kafka-console-producer.sh --broker-list node01:9092,node01:9092,node01:9092 --topic spark_kafka 
 
代码演示 
import org.apache.kafka.clients.consumer.ConsumerRecord 
import org.apache.kafka.common.serialization.StringDeserializer 
import org.apache.spark.streaming.dstream.{DStream, InputDStream} 
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies} 
import org.apache.spark.streaming.{Seconds, StreamingContext} 
import org.apache.spark.{SparkConf, SparkContext} 
 
object SparkKafkaDemo { 
def main(args: Array[String]): Unit = { 
//1.创建StreamingContext 
//spark.master should be set as local[n], n > 1 
val conf = new SparkConf.setAppName("wc").setMaster("local[*]") 
val sc = new SparkContext(conf) 
sc.setLogLevel("WARN") 
val ssc = new StreamingContext(sc,Seconds(5))//5表示5秒中对数据进行切分形成一个RDD 
//准备连接Kafka的参数 
val kafkaParams = Map[String, Object]( 
"bootstrap.servers" -> "node01:9092,node02:9092,node03:9092", 
"key.deserializer" -> classOf[StringDeserializer], 
"value.deserializer" -> classOf[StringDeserializer], 
"group.id" -> "SparkKafkaDemo", 
//earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
//latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 
//none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 
//这里配置latest自动重置偏移量为最新的偏移量,即如果有偏移量从偏移量位置开始消费,没有偏移量从新来的数据开始消费 
"auto.offset.reset" -> "latest", 
//false表示关闭自动提交.由spark帮你提交到Checkpoint或程序员手动维护 
"enable.auto.commit" -> (false: java.lang.Boolean) 
) 
val topics = Array("spark_kafka") 
//2.使用KafkaUtil连接Kafak获取数据 
val recordDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc, 
LocationStrategies.PreferConsistent,//位置策略,源码强烈推荐使用该策略,会让Spark的Executor和Kafka的Broker均匀对应 
ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))//消费策略,源码强烈推荐使用该策略 
//3.获取VALUE数据 
val lineDStream: DStream[String] = recordDStream.map(_.value)//_指的是ConsumerRecord 
val wrodDStream: DStream[String] = lineDStream.flatMap(_.split(" ")) //_指的是发过来的value,即一行数据 
val wordAndOneDStream: DStream[(String, Int)] = wrodDStream.map((_,1)) 
val result: DStream[(String, Int)] = wordAndOneDStream.reduceByKey(_+_) 
result.print 
ssc.start//开启 
ssc.awaitTermination//等待优雅停止 
} 
} 
 
好了，本篇主要讲解的 SparkStreaming 整合 Kafka 的过程，并带大家复习了一波Kafka的基础知识，如果对你有用的话，麻烦动手手点个“在看”吧~ 
   
本文由作者首发 CSDN 博客，原文链接： 
   
☞开源激荡 30 年：从免费社区到价值数十亿美元公司 
☞理解 AI 最伟大的成就之一：卷积神经网络的局限性 
☞GitHub 标星 10,000+，Apache 顶级项目 ShardingSphere 的开源之路 
☞港科大郑光廷院士问诊未来，揭露 AI 最新应用与实践 
☞大促下的智能运维挑战：阿里如何抗住“双11猫晚”？ 
☞以太坊2.0中的Custody Game及MPC实现 
☞很用心的为你写了9道MySQL面试题，建议收藏！