1. 回顾 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 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用与大数据实时处理领域。

• 消息队列：Kafka 本质上是一个 MQ（Message Queue），使用消息队列的好处？（面试会问）：
  • 解耦：允许我们独立的扩展或修改队列两边的处理过程；
  • 可恢复性：即使一个处理消息的进程挂掉，加入队列中的消息仍可以在系统恢复后被处理；
  • 缓冲：有助于解决生产消息和消费消息的处理速度不一致的情况；
  • 灵活性&峰值处理能力：不会因为突发的超负荷的请求而完全崩溃，消息队列能够使关键组件顶住突发的访问压力；
  • 异步通信：消息队列允许用户把消息放入队列但不立即处理它；
• 发布/订阅模式：

一对多，生产者将消息发布到 Topic 中，有多个消费者订阅该主题，发布到 Topic 的消息会被所有订阅者消费，被消费的数据不会立即从 Topic 清除。
Apache Kafka： 最原始功能【消息队列】，缓冲数据，具有发布订阅功能（类似微信公众号）。

kafka文档

• Kafka Connect:
  connect
  connect-jdbc

• Kafka Stream:
  streams
  streaming-kafka-integration

• kafka monitor:
  monitoring
  KafkaOffsetMonitor
  kafka-manager

• kafka生态圈：
  Ecosystem

• SparkStreaming整合Kafka
  streaming-kafka-integration

2. 集成方式

Spark Streaming与Kafka集成，有两套API，原因在于Kafka Consumer API有两套，文档：http://spark.apache.org/docs/2.4.5/streaming-kafka-integration.html。

• 方式一：Kafka 0.8.x版本
  • 老的Old Kafka Consumer API
  • 文档：streaming-kafka-0-8-integration
  • 老的Old消费者API，有两种方式：
    • 第一种：高级消费API（Consumer High Level API），Receiver接收器接收数据
    • 第二种：简单消费者API(Consumer Simple Level API) ，Direct 直接拉取数据
• 方式二：Kafka 0.10.x版本
  • 新的 New Kafka Consumer API
  • 文档：streaming-kafka-0-10-integration
  • 核心API：KafkaConsumer、ConsumerRecorder
    

3. 两种方式区别

使用Kafka Old Consumer API集成两种方式，虽然实际生产环境使用Direct方式获取数据，但是在面试的时候常常问到两者区别。
文档：http://spark.apache.org/docs/2.4.5/streaming-kafka-0-8-integration.html

• Receiver-based Approach：
  • 基于接收器方式，消费Kafka Topic数据，但是企业中基本上不再使用；
  • Receiver作为常驻的Task运行在Executor等待数据，但是一个Receiver效率低，需要开启多个，再手动合并数据(union)，再进行处理，很麻烦；
  • Receiver那台机器挂了，可能会丢失数据，所以需要开启WAL(预写日志)保证数据安全，那么效率又会降低；
  • Receiver方式是通过zookeeper来连接kafka队列，调用Kafka高阶API，offset存储在zookeeper，由Receiver维护；
  • Spark在消费的时候为了保证数据不丢也会在Checkpoint中存一份offset，可能会出现数据不一致；
• Direct Approach (No Receivers)：
  • 直接方式，Streaming中每批次的每个job直接调用Simple Consumer API获取对应Topic数据，此种方式使用最多；
  • Direct方式是直接连接kafka分区来获取数据，从每个分区直接读取数据大大提高并行能力
  • Direct方式调用Kafka低阶API(底层API)，offset自己存储和维护，默认由Spark维护在checkpoint中，消除了与zk不一致的情况 ；
  • 当然也可以自己手动维护，把offset存在MySQL、Redis和Zookeeper中；

上述两种方式区别，如下图所示：

4. 整合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	at most once 最多被处理一次	会丢失数据
Receiver+WAL	at least once 最少被处理一次	不会丢失数据，但可能会重复消费，且效率低
Direct+手动操作	exactly once 只被处理一次/精准一次	不会丢失数据，也不会重复消费，且效率高

• 注意:
  开发中SparkStreaming和kafka集成有两个版本：0.8及0.10+
  0.8版本有Receiver（zookeeper）和Direct（topic）模式(但是0.8版本生产环境问题较多，在Spark2.3之后不支持0.8版本了)
  0.10以后只保留了direct模式(Reveiver模式不适合生产环境)，并且0.10版本API有变化(更加强大)
  streaming-kafka-integration

5. [了解]Apache Kafka数据源0.8版本对接方式

kafka作为一个实时的分布式消息队列，实时的生产和消费消息，这里我们可以利用SparkStreaming实时地读取kafka中的数据，然后进行相关计算。
在Spark1.3版本后，KafkaUtils里面提供了两个创建dstream的方法，一种为KafkaUtils.createDstream，另一种为KafkaUtils.createDirectStream。
官方文档说明
streaming-kafka-integration

5.1 回顾kafka的安装与使用

参考前面kafka安装部署

5.1.1 三台机器安装zookeeper

5.1.2 三台机器安装kafka集群

5.1.2.1 下载kafka安装压缩包

kafka

5.1.2.2 上传压缩包并解压

将kafka的安装包上传到第一台服务器的/export/softwares路径下面去，然后解压到/export/servers
这里统一使用kafka_2.11-1.0.0.tgz 这个版本

5.1.2.3 修改kafka配置文件

第一台机器修改kafka配置文件server.properties

broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/export/server/kafka/kafka-logs
num.partitions=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=node1:2181,node2:2181,node3:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=node1

第二台机器修改kafka配置文件server.properties

broker.id=1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/export/server/kafka/kafka-logs
num.partitions=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=node1:2181,node2:2181,node3:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=node2

第三台机器修改kafka配置文件server.properties

broker.id=2
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/export/server/kafka/kafka-logs
num.partitions=2
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=node1:2181,node2:2181,node3:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=node3

5.1.2.4 启动kafka集群

三台机器启动kafka服务

bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 &

后台启动命令

5.1.3 kafka的命令行的管理使用

#查看topic信息
kafka-topics.sh --list --zookeeper node1:2181,node2:2181,node3:2181
#创建topic
kafka-topics.sh --create --partitions 3 --replication-factor 2 --topic kafkatopic --zookeeper node1:2181,node2:2181,node3:2181 
#模拟生产者
kafka-console-producer.sh --broker-list node1:2181,node2:2181,node3:2181 --topic kafkatopic
#模拟消费者
kafka-console-consumer.sh --from-beginning --topic kafkatopic --zookeeper node1:2181,node2:2181,node3:2181

5.2 第一种方式对接kafka之CreateDstream方式

KafkaUtils.createDstream(ssc, [zk], [group id], [per-topic,partitions] ) 使用了receivers接收器来接收数据，利用的是Kafka高层次的消费者api，对于所有的receivers接收到的数据将会保存在Spark executors中，然后通过Spark Streaming启动job来处理这些数据，默认会丢失，可启用WAL日志，它同步将接受到数据保存到分布式文件系统上比如HDFS。 所以数据在出错的情况下可以恢复出来 。

A、创建一个receiver接收器来对kafka进行定时拉取数据，这里产生的dstream中rdd分区和kafka的topic分区不是一个概念，故如果增加特定topic的分区数仅仅是增加一个receiver中消费topic的线程数，并没有增加spark的并行处理的数据量。
B、对于不同的group和topic可以使用多个receivers创建不同的DStream
C、如果启用了WAL(spark.streaming.receiver.writeAheadLog.enable=true)
同时需要设置存储级别(默认StorageLevel.MEMORY_AND_DISK_SER_2)

5.2.1 第一步：导入jar包

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>2.4.5</version>
</dependency>

5.2.2 第二步：创建kafka的topic

node01服务器执行以下命令创建kafka的topic sparkafka

# 1.启动Zookeeper 服务
zookeeper-daemon.sh start
# 2.启动Kafka 服务
kafka-daemon.sh start
# 3.Create Topic
cd /export/servers/kafka/
bin/kafka-topics.sh --create --topic sparkafka  \
--partitions 3 --replication-factor 1 --zookeeper node1:2181/kafka200

5.2.3 第三步：使用脚本启动kafka生产者

node01服务器执行以下命令通过脚本模拟kafka生产者

#4 Producer
cd /export/servers/kafka_2.11-1.0.0/
bin/kafka-console-producer.sh --topic sparkafka --broker-list node1:9092

开启消费者

#5 Consumer
kafka-console-consumer.sh --topic wc-topic \
--bootstrap-server node1:9092 --from-beginning

5.2.4 第四步：开发SparkStreaming对接kafka代码

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import scala.collection.immutableobject StreamKafkaReceiver {def main(args: Array[String]): Unit = {//1、创建sparkConfval sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingKafka_Receiver").setMaster("local[4]").set("spark.streaming.receiver.writeAheadLog.enable", "true") //开启wal预写日志，保存数据源的可靠性//2、创建sparkContextval sc = new SparkContext(sparkConf)sc.setLogLevel("WARN")//3、创建StreamingContextval ssc = new StreamingContext(sc, Seconds(5))//设置checkpointssc.checkpoint("./Kafka_Receiver")//4、定义zk地址val zkQuorum = "node01:2181,node02:2181,node03:2181"//5、定义消费者组val groupId = "sparkafka_group"//6、定义topic相关信息 Map[String, Int]// 指定消费的topic的名称和消费topic的线程数val topics = Map("sparkafka" -> 3)//7、通过KafkaUtils.createDStream对接kafka//这个时候相当于同时开启3个receiver接受数据val receiverDstream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {val stream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)stream})//使用ssc.union方法合并所有的receiver中的数据val unionDStream: DStream[(String, String)] = ssc.union(receiverDstream)//8、获取topic中的数据val topicData: DStream[String] = unionDStream.map(_._2)//9、切分每一行,每个单词计为1val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_, 1))//10、相同单词出现的次数累加val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _)//11、打印输出result.print()//开启计算ssc.start()ssc.awaitTermination()}
}

5.3 第二种方式对接kafka之CreateDirectStream方式

使用kafka的低阶API进行消费，消费数据的offset全部维护在kafka当中的一个topic当中，会自动提交offset，同时也可以手动提交维护offset更加安全

import kafka.serializer.StringDecoder
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils//todo:利用sparkStreaming对接kafka实现单词计数----采用Direct(低级API)
object SparkStreamingKafka_Direct {def main(args: Array[String]): Unit = {//1、创建sparkConfval sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingKafka_Direct").setMaster("local[2]")//2、创建sparkContextval sc = new SparkContext(sparkConf)sc.setLogLevel("WARN")//3、创建StreamingContextval ssc = new StreamingContext(sc, Seconds(5))ssc.checkpoint("./Kafka_Direct")//4、配置kafka相关参数val kafkaParams = Map("metadata.broker.list" -> "node01:9092,node02:9092,node03:9092", "group.id" -> "Kafka_Direct")//5、定义topicval topics = Set("sparkafka")//6、通过 KafkaUtils.createDirectStream接受kafka数据，这里采用是kafka低级api偏移量不受zk管理val dstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)//7、获取kafka中topic中的数据val topicData: DStream[String] = dstream.map(_._2)//8、切分每一行,每个单词计为1val wordAndOne: DStream[(String, Int)] = topicData.flatMap(_.split(" ")).map((_, 1))//9、相同单词出现的次数累加val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _)//10、打印输出result.print()//开启计算ssc.start()ssc.awaitTermination()}
}

6. [掌握]Apche kafka数据源0.10版本对接

手动提交offset，进行offset的管理维护，保证数据不会丢失，推荐使用

• 说明
  spark-streaming-kafka-0-10版本中，API有变化，操作更加灵活，开发中使用
• API
  streaming-kafka-0-10-integration

6.1 环境准备

• node01服务器执行以下命令创建kafka的topic sparkafka

# 1.启动Zookeeper 服务
zookeeper-daemon.sh start
# 2.启动Kafka 服务
kafka-daemon.sh start
# 3.Create Topic
cd /export/servers/kafka/
bin/kafka-topics.sh --create --topic sparkafka  \
--partitions 3 --replication-factor 1 --zookeeper node1:2181/kafka200

• 使用脚本启动kafka生产者
  node01服务器执行以下命令通过脚本模拟kafka生产者

#4 Producer
cd /export/servers/kafka_2.11-1.0.0/
bin/kafka-console-producer.sh --topic sparkafka --broker-list node1:9092

• 开启消费者

#5 Consumer
kafka-console-consumer.sh --topic wc-topic \
--bootstrap-server node1:9092 --from-beginning

6.2 导入对应jar包

<!-- <dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>2.4.5</version>
</dependency>-->
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.11</artifactId><version>2.4.5</version>
</dependency>

6.3 Kafka整合实现WordCount

import java.langimport 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, ConsumerStrategy, KafkaUtils, LocationStrategies, LocationStrategy}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}/*** DESC:* Complete data processing and modeling process steps:*/
object kafka_010_testWordCount {var kafkaParams = Map[String, Object](//这里指的是broker的地址"bootstrap.servers" -> "node01:9092,node02:9092,node03:9092","key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> "sparkkafkagroup",//这里的auto.offset.reset代表的是自动重置offset为latest就表示的是最新的偏移量，如果没有偏移从最新的位置开始"auto.offset.reset" -> "latest",//false表示的是手动提交offset，如果为true代表的是自动提交offset//这里如果是false手动提交，默认由SparkStreaming提交到checkpoint中，在这里也可以根据用户或程序员将offset偏移量提交到mysql或redis中"enable.auto.commit" -> (false: lang.Boolean))def main(args: Array[String]): Unit = {//    * 1-准备环境val conf: SparkConf = new SparkConf().setAppName("kafka_010_SparkStreaming").setMaster("local[*]")val sc = new SparkContext(conf)sc.setLogLevel("WARN")val streamingContext = new StreamingContext(sc, Seconds(5))//    * 2-准备Kafka的各种配置//    * 3-使用Kafka的Utils的类实现lowlevelapi的DirectDStream方式//    ssc: StreamingContext,//    locationStrategy: LocationStrategy,//    consumerStrategy: ConsumerStrategy[K, V]val recordStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](streamingContext,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](Array("kafkatopic"), kafkaParams))//    * 4-数据处理--首先从当前的数据中获取对应的value的值---这里实现基础的wordCount操作---flatMap---map---reduceBykey或updateStateBykey//(topic = kafkatopic, partition = 2, offset = 5, CreateTime = 1585120000150, checksum = 2539994510, serialized key size = -1, serialized value size = 11, key = null, value = hello flink))val value: DStream[String] = recordStream.map(_.value()) //获取当前数据的值，将对应的值取出来在进行对应的wordcount的操作//    * 5-打印结果val result: DStream[(String, Int)] = value.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)result.print()//    * 6-开启程序streamingContext.start()//    * 7-等待用户关闭streamingContext.awaitTermination()}
}

6.4 Kafka整合代码获取偏移量

手动提交offset管理，保证数据消费exactly once

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object StreamingKafka1 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[4]").setAppName("NetworkWordCount")val context = new SparkContext(conf)context.setLogLevel("WARN")val ssc = new StreamingContext(context, Seconds(1))//创建topicval brokers= "node01:9092,node02:9092,node03:9092"val sourcetopic="kafkatopic";//创建消费者组var group="sparkafkaGroup"//消费者配置val kafkaParam = Map("bootstrap.servers" -> brokers,//用于初始化链接到集群的地址"key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],//用于标识这个消费者属于哪个消费团体"group.id" -> group,//如果没有初始化偏移量或者当前的偏移量不存在任何服务器上，可以使用这个配置属性//可以使用这个配置，latest自动重置偏移量为最新的偏移量"auto.offset.reset" -> "latest",//如果是true，则这个消费者的偏移量会在后台自动提交"enable.auto.commit" -> (false: java.lang.Boolean));var stream = KafkaUtils.createDirectStream[String,String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String,String](Array("kafkatopic"),kafkaParam))//循环遍历每个RDD当中的数据stream.foreachRDD(f =>{//判断如果rdd当中有数据，就进行处理，没有数据就不用处理if(f.count() > 0){println("接收kafka当中的数据")//每个分区当中的数据进行循环遍历，遍历每个分区当中每一行的数据f.foreach(f =>{//获取kafka当中的数据内容
print(“record topic is:”,f)val kafkaValue: String = f.value()println(kafkaValue)})//打印offset的信息val offsetRanges: Array[OffsetRange] = f.asInstanceOf[HasOffsetRanges].offsetRangesf.foreachPartition { iter =>val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)println(s"$topic{o.topic} $partition{o.partition} $fromOffset{o.fromOffset} $untilOffset{o.untilOffset}")}println("=============================")// 等输出操作完成后提交offsetstream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)}})ssc.start()ssc.awaitTermination()}
}

结果：

这里面为什么是三个分区是因为创建的时候使用的是3个分区
上述代码提交offset默认到checkpoint中