首页 > 其他分享> > 2021-02-25 大数据课程笔记 day36

2021-02-25 大数据课程笔记 day36

2021-02-25 18:58:07 作者：互联网

时间煮雨
@R星校长

Spark第八天【SparkStreaming内容】

主要内容

Kafka应用场景
Kafka分布式消息系统原理
Kafka集群搭建
Kafka 操作命令
Kafka Leader 均衡机制
SparkStreaming与Kafka整合

学习目标

第一节 Kafka 介绍

1. kafka是什么？使用场景？

Kafka是一个高吞吐的分布式消息队列系统。特点是生产者消费者模式，先进先出（FIFO）保证顺序，自己不丢数据，默认每隔7天清理数据。消息列队常见场景：系统之间解耦合、峰值压力缓冲、异步通信。

2. kafka生产消息、存储消息、消费消息 在这里插入图片描述
Kafka架构是由producer（消息生产者）、consumer（消息消费者）、borker(kafka集群的server，负责处理消息读、写请求，存储消息，在kafka cluster这一层这里，其实里面是有很多个broker)、topic（消息队列/分类相当于队列，里面有生产者和消费者模型）、zookeeper(元数据信息存在zookeeper中，包括：存储消费偏移量，topic话题信息，partition信息) 这些部分组成。

kafka里面的消息是有topic来组织的，简单的我们可以想象为一个队列，一个队列就是一个topic，然后它把每个topic又分为很多个partition，这个是为了做并行的，在每个partition内部消息强有序，相当于有序的队列，其中每个消息都有个序号offset，比如0到12，从前面读往后面写。一个partition对应一个broker，一个broker可以管多个partition，比如说，topic有6个partition，有两个broker，那每个broker就管3个partition。这个partition可以很简单想象为一个文件，当数据发过来的时候它就往这个partition上面append，追加就行，消息不经过内存缓冲，直接写入文件，kafka和很多消息系统不一样，很多消息系统是消费完了我就把它删掉，而kafka是根据时间策略删除，而不是消费完就删除，在kafka里面没有一个消费完这么个概念，只有过期这样一个概念。

producer自己决定往哪个partition里面去写，这里有一些的策略，譬如hash。consumer自己维护消费到哪个offset，每个consumer都有对应的group，group内是queue消费模型（各个consumer消费不同的partition，因此一个消息在group内只消费一次），group间是publish-subscribe消费模型，各个group各自独立消费，互不影响，因此一个消息在被每个group消费一次。

Kafka集群运行时，是有一个Controller控制器来管理集群，这个控制器主要负责Kafka集群节点管理，分区重分配，topic管理，负责分配topic每个分区由哪个broker读写和存储。Kafka集群中每个broker节点都是平等机会争取这个控制器，控制器出现是避免每个broker节点都与zookeeper来进行大量通信带来性能开销，zookeepeer中的元数据信息也是通过Controller汇报来的。当控制器失败时，其他broker节点会由zookeeper协调选择成对应的新的Controller。

3. kafka的特点

系统的特点：生产者消费者模型，FIFO
Partition内部是FIFO的，partition之间呢不是FIFO的，当然我们可以把topic设为一个partition，这样就是严格的FIFO。
高性能：单节点支持上千个客户端，百MB/s吞吐，接近网卡的极限
持久性：消息直接持久化在普通磁盘上且性能好
直接写到磁盘中去，就是直接append到磁盘里去，这样的好处是直接持久化，数据不会丢失，第二个好处是顺序写，然后消费数据也是顺序的读，所以持久化的同时还能保证顺序，比较好，因为磁盘顺序读比较好。
分布式：数据副本冗余、流量负载均衡、可扩展
分布式，数据副本，也就是同一份数据可以到不同的broker上面去，也就是当一份数据，磁盘坏掉的时候，数据不会丢失，比如3个副本，就是在3个机器磁盘都坏掉的情况下数据才会丢，在大量使用情况下看这样是非常好的，负载均衡，可扩展，在线扩展，不需要停服务。
很灵活：消息长时间持久化+Client维护消费状态
消费方式非常灵活，第一原因是消息持久化时间跨度比较长，一天或者一星期等，第二消费状态自己维护消费到哪个地方了可以自定义消费偏移量。

第二节 Kafka使用

1. kafka集群搭建

1) 上传kafka_2.10-0.8.2.2.tgz包到三个不同节点上，解压。

2) 配置…/ kafka_2.10-0.8.2.2/config/server.properties文件
节点编号：（不同节点按0,1,2,3整数来配置）在这里插入图片描述
真实数据存储位置：
zookeeper的节点：
3) 启动zookeeper集群。

4) 三个节点上，启动kafka:

bin/kafka-server-start.sh   config/server.properties

最好使用自己写的脚本启动，将启动命令写入到一个文件：

nohup bin/kafka-server-start.sh   config/server.properties > kafka.log 2>&1 &

脚本附件：
（放在与bin同一级别下，注意创建后要修改权限：chmod 755 startkafka.sh）
在这里插入图片描述
5) 相关命令：

创建topic：

./kafka-topics.sh --zookeeper node3:2181,node4:2181,node5:2181  
--create --topic topic2017 --partitions 3 --replication-factor 3

用一台节点控制台来当kafka的生产者：

./kafka-console-producer.sh  --topic  topic2017 
--broker-list node1:9092,node2:9092,node3:9092

用另一台节点控制台来当kafka的消费者：

./kafka-console-consumer.sh --zookeeper node3:2181,node4:2181,node5:2181 --topic topic2017

查看kafka中topic列表:

./kafka-topics.sh  --list --zookeeper node3:2181,node4:2181,node5:2181

查看kafka中topic的描述：

./kafka-topics.sh --describe --zookeeper node3:2181,node4:2181,node5:2181  --topic topic2017

在这里插入图片描述
注意：ISR是检查数据的完整性有哪些个节点。

查看zookeeper中topic相关信息：
启动zookeeper客户端：
./zkCli.sh
查看topic相关信息：
ls /brokers/topics/
查看消费者相关信息：
ls /consumers

6) 删除kafka中的数据。

① ：在kafka集群中删除topic，当前topic被标记成删除。

./kafka-topics.sh --zookeeper node3:2181,node4:2181,node5:2181 --delete --topic t1205

在每台broker节点上删除当前这个topic对应的真实数据。

② ：进入zookeeper客户端，删除topic信息

rmr /brokers/topics/t1205

③ ：删除zookeeper中被标记为删除的topic信息

rmr /admin/delete_topics/t1205

以上方式比较笨重，可以直接在每台broker节点中的../config/server.properties中配置属性：delete.topic.enable=true,重启kafka集群即可实现删除topic时，自动清除topic信息。

2. kafka的leader的均衡机制

当一个broker停止或者crashes时，所有本来将它作为leader的分区将会把leader转移到其他broker上去，极端情况下，会导致同一个leader管理多个分区，导致负载不均衡，同时当这个broker重启时，如果这个broker不再是任何分区的leader,kafka的client也不会从这个broker来读取消息，从而导致资源的浪费。

kafka中有一个被称为优先副本（preferred replicas）的概念。如果一个分区有3个副本，且这3个副本的优先级别分别为0,1,2，根据优先副本的概念，0会作为leader 。当0节点的broker挂掉时，会启动1这个节点broker当做leader。当0节点的broker再次启动后，会自动恢复为此partition的leader。不会导致负载不均衡和资源浪费，这就是leader的均衡机制。

在配置文件conf/ server.properties中配置开启（默认就是开启）：

auto.leader.rebalance.enable true

在Kafka0.11版本中，Leader均衡机制需要手动执行命令触发，首先需要配置如下json文件：

topic.json:
{
 "partitions":
  [
    {"topic": "topicName", "partition": 0},
    {"topic": "topicName", "partition": 1},
    {"topic": "topicName", "partition": 2}
  ]
}

手动执行如下命令，进行Leader均衡：

./kafka-preferred-replica-election.sh --zookeeper mynode3:2181,mynode4:2181,mynode5:2181 --path-to-json-file topic.json

3. kafka 0.11版本改变

kafka 0.8.2版本消费者offset存储在zookeeper中，对于zookeeper而言每次写操作代价是很昂贵的，而且zookeeper集群是不能扩展写能力。kafka 0.11版本默认使用新的消费者api ,消费者offset会更新到一个kafka自带的topic【__consumer_offsets】中。以消费者组groupid 为单位，可以查询每个组的消费topic情况：

#查看所有消费者组
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092 --list

#查看消费者消费的offset位置信息
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092  --describe --group MyGroupId

#重置消费者组的消费offset信息 ，--reset-offsets –all-topics 所有offset。--to-earliest 最小位置。
# --execute 执行
./kafka-consumer-groups.sh --bootstrap-server c7node1:9092,c7node2:9092,c7node3:9092 --group MyGroupId --reset-offsets --all-topics --to-earliest --execute

第三节 Receiver模式（1.6版本）

SparkStreaming1.6版本+Kafka0.8.2 版本整合方式有两种模式，一种是Receiver模式，另一种是Direct模式。

Receiver模式原理图
Receiver模式理解

Receiver模式中，SparkStreaming使用Receiver接收器模式来接收kafka中的数据，即会将每批次数据都存储在Spark端，默认的存储级别为MEMORY_AND_DISK_SER_2，从Kafka接收过来数据之后，还会将数据备份到其他Executor节点上，当完成备份之后，再将消费者offset数据写往zookeeper中，然后再向Driver汇报数据位置，Driver发送task到数据所在节点处理数据。

这种模式使用zookeeper来保存消费者offset，等到SparkStreaming重启后，从zookeeper中获取offset继续消费。

当Driver挂掉时，同时消费数据的offset已经更新到zookeeper中时，SparkStreaming重启后，接着zookeeper存储的offset继续处理数据，这样就存在丢失数据的问题。

为了解决以上丢失数据的问题，可以开启WAL(write ahead log)预写日志机制，将从kafka中接收来的数据备份完成之后，向指定的checkpoint中也保存一份，这样当SparkStreaming挂掉，重新启动再处理数据时，会处理Checkpoint中最近批次的数据，将消费者offset继续更新保存到zookeeper中。

开启WAL机制，需要设置checpoint,由于一般checkpoint路径都会设置到HDFS中，HDFS本身会有副本，所以这里如果开启WAL机制之后，可以将接收数据的存储级别降级，去掉“_2”级别。

开启WAL机制之后带来了新的问题：

1) 数据重复处理问题

由于开启WAL机制，会处理checkpoint中最近一段时间批次数据，这样会造成重复处理数据问题。所以对于数据需要精准消费的场景，不能使用receiver模式。如果不开启WAL机制Receiver模式有丢失数据的问题，开启WAL机制之后有重复处理数据的问题，对于精准消费数据的场景，只能人为保存offset来保证数据消费的准确性。

2) 数据处理延迟加大问题

数据在节点之间备份完成后再向checkpoint中备份，之后再向Zookeeper汇报数据offset，向Driver汇报数据位置，然后Driver发送task处理数据。这样加大了数据处理过程中的延迟。

对于精准消费数据的问题，需要我们从每批次中获取offset然后保存到外部的数据库来实现来实现仅一次消费数据。但是Receiver模式底层读取Kafka数据的实现使用的是High Level Consumer Api，这种Api不支持获取每批次消费数据的offset。所以对于精准消费数据的场景不能使用这种模式。

Receiver模式总结

1) Receiver模式采用了Receiver接收器的模式接收数据。会将每批次的数据存储在Executor内存或者磁盘中。
2) Receiver模式有丢失数据问题，开启WAL机制解决，但是带来新的问题。
3) receiver模式依赖zookeeper管理消费者offset。
4) SparkStreaming读取Kafka数据，相当于Kafka的消费者，底层读取Kafka采用了“High Level Consumer API”实现，这种api没有提供操作每批次数据offset的接口，所以对于精准消费数据的场景想要人为控制offset是不可能的。

Receiver模式代码

1.	SparkConf conf = new SparkConf()
2.	.setAppName("SparkStreamingOnKafkaReceiver").setMaster("local[2]");
3.	//开启预写日志 WAL机制
4.	conf.set("spark.streaming.receiver.writeAheadLog.enable","true");
5.	
6.	JavaStreamingContext jsc = new JavaStreamingContext(conf,     Durations.seconds(5));
7.	jsc.checkpoint("./receivedata");
8.	
9.	Map<String, Integer> topicConsumerConcurrency = new HashMap<String, Integer>();
10.	/**
11.	* 设置读取的topic和接受数据的线程数
12.	*/
13.	topicConsumerConcurrency.put("topic1101",1);
14.	
15.	/**
16.	* 第一个参数是StreamingContext
17.	* 第二个参数是ZooKeeper集群信息（接受Kafka数据的时候会从Zookeeper中获得Offset等元数据信息）
18.	* 第三个参数是Consumer Group 消费者组
19.	* 第四个参数是消费的Topic以及并发读取Topic中Partition的线程数
20.	* 
21.	* 注意：
22.	* KafkaUtils.createStream 使用五个参数的方法，设置receiver的存储级别
23.	*/
24.	JavaPairReceiverInputDStream<String,String> lines = KafkaUtils.createStream(
25.	     jsc,
26.	     "mynode3:2181,mynode4:2181,mynode5:2181",
27.	     "MyFirstConsumerGroup", 
28.	     topicConsumerConcurrency/*,StorageLevel.MEMORY_AND_DISK()*/
29.	);
30.	
31.	JavaDStream<String> words = lines.flatMap(new   FlatMapFunction<Tuple2<String,String>, String>(){ 
32.	  public Iterable<String> call(Tuple2<String,String> tuple) throws Exception {
33.	    System.out.println("key = "+tuple._1);
34.	    System.out.println("value = "+tuple._2);
35.	    return Arrays.asList(tuple._2.split("\t"));
36.	  }
37.	});
38.	
39.	
40.	JavaPairDStream<String, Integer> pairs = 
41.	words.mapToPair(new PairFunction<String, String, Integer>() {
42.	  public Tuple2<String, Integer> call(String word) throws Exception {
43.	    return new Tuple2<String, Integer>(word, 1);
44.	  }
45.	});
46.	
47.	
48.	JavaPairDStream<String, Integer> wordsCount = 
49.	pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { 
50.	  public Integer call(Integer v1, Integer v2) throws Exception {
51.	    return v1 + v2;
52.	  }
53.	});
54.	
55.	
56.	wordsCount.print(100);
57.	
58.	
59.	jsc.start();
60.	jsc.awaitTermination();
61.	jsc.close();
62.

Receiver的并行度设置

receiver的并行度是由spark.streaming.blockInterval来决定的，默认为200ms,假设batchInterval为5s,那么每隔blockInterval就会产生一个block,这里就对应每批次产生RDD的partition,这样5秒产生的这个Dstream中的这个RDD的partition为25个，并行度就是25。如果想提高并行度可以减少blockInterval的数值，但是最好不要低于50ms。

第四节 Direct模式（1.6版本）

SparkStreaming1.6 + kafka0.8.2版本整合方式中，除了有Receiver模式之外，另一种整合方式就是Driect模式。

Direct模式原理图
Direct模式理解

Driect模式就是将kafka看成存数据的一方，这种模式没有采用Receiver接收器模式，而是采用直连的方式，不是被动接收数据，而是主动去取数据，在处理每批次时会根据offset位置向Kafka中获取数据。

消费者偏移量也不是用zookeeper来管理，而是使用Spark自己进行消费者偏移量的维护，如果没有设置checkpoint机制，消费偏移量默认是存在内存中，如果设置了checkpoint目录，那么消费偏移量也会保存在checkpoint中。当SparkStreaming停止后，我们可以使用val ssc = StreamFactory.getOrCreate(checkpointDir,Fun)来恢复停止之前SparkStreaming处理数据的进度，当然，这种方式存在重复消费数据和逻辑改变之后不可执行的问题。

Direct模式底层读取Kafka数据实现是Simple Consumer api实现，这种api提供了从每批次数据中获取offset的接口，所以对于精准消费数据的场景，可以使用Direct 模式手动维护offset方式来实现数据精准消费。

Direct模式并行度设置
Direct模式的并行度与当前读取的topic的partition个数一致，所以Direct模式并行度由读取的kafka中topic的partition数决定的。
Direct模式代码

1.	SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkStreamingOnKafkaDirected");
2.	JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));
3.	/**
4.	* 可以不设置checkpoint 不设置不保存offset,offset默认在内存中有一份，如果设置checkpoint在checkpoint也有一份offset， 一般要设置。
5.	*/
6.	jsc.checkpoint("./checkpoint");
7.	Map<String, String> kafkaParameters = new HashMap<String, String>();
8.	kafkaParameters.put("metadata.broker.list", "mynode1:9092,mynode2:9092,mynode3:9092");
9.	HashSet<String> topics = new HashSet<String>();
10.	topics.add("topic1101");
11.	
12.	JavaPairInputDStream<String,String> lines =    KafkaUtils.createDirectStream(jsc,
13.	    String.class, 
14.	    String.class,
15.	    StringDecoder.class,
16.	    StringDecoder.class,
17.	    kafkaParameters,
18.	  topics);
19.	
20.	JavaDStream<String> words = lines.flatMap(new    FlatMapFunction<Tuple2<String,String>, String>() {
21.	  public Iterable<String> call(Tuple2<String,String> tuple) throws Exception {
22.	    System.out.println("key = "+tuple._1);
23.	    return Arrays.asList(tuple._2.split("\t"));
24.	  }
25.	});
26.	
27.	JavaPairDStream<String, Integer> pairs = words.mapToPair(new   PairFunction<String, String, Integer>() {
28.	  public Tuple2<String, Integer> call(String word) throws Exception   {
29.	    return new Tuple2<String, Integer>(word, 1);
30.	  }
31.	});
32.	
33.	
34.	JavaPairDStream<String, Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
35.	  public Integer call(Integer v1, Integer v2) throws Exception {
36.	    return v1 + v2;
37.	  }
38.	});
39.	
40.	wordsCount.print();
41.	jsc.start();
42.	jsc.awaitTermination();
43.	jsc.close();

第五节 SparkStreaming2.3 Direct模式

1. SparkStreaming2.3+kafka 改变
1) 丢弃了SparkStreaming+kafka 的receiver模式。

2) 采用了新的消费者api实现，类似于1.6中SparkStreaming 读取 kafka Direct模式。并行度一样。

3) 因为采用了新的消费者api实现，所以相对于1.6的Direct模式【simple api实现】，api使用上有很大差别。未来这种api有可能继续变化。

4) kafka中有两个参数：

heartbeat.interval.ms：这个值代表 kafka集群与消费者之间的心跳间隔时间，kafka 集群确保消费者保持连接的心跳通信时间间隔。这个时间默认是3s.这个值必须设置的比session.timeout.ms appropriately 小，一般设置不大于 session.timeout.ms的1/3。

session.timeout.ms：这个值代表消费者与kafka之间的session 会话超时时间，如果在这个时间内，kafka 没有接收到消费者的心跳【heartbeat.interval.ms 控制】，那么kafka将移除当前的消费者。这个时间默认是10s。这个时间是位于 group.min.session.timeout.ms【6s】和 group.max.session.timeout.ms【300s】之间的一个参数,如果SparkSteaming 批次间隔时间大于5分钟，也就是大于300s,那么就要相应的调大group.max.session.timeout.ms 这个值。

5) 大多数情况下，SparkStreaming读取数据使用 LocationStrategies.PreferConsistent 这种策略，这种策略会将分区均匀的分布在集群的Executor之间。

如果Executor在kafka 集群中的某些节点上，可以使用 LocationStrategies.PreferBrokers 这种策略，那么当前这个Executor 中的数据会来自当前broker节点。

如果节点之间的分区有明显的分布不均，可以使用 LocationStrategies.PreferFixed 这种策略,可以通过一个map 指定将topic分区分布在哪些节点中。

6) 新的消费者api 可以将kafka 中的消息预读取到缓存区中，默认大小为64k。默认缓存区在 Executor 中，加快处理数据速度。可以通过参数 spark.streaming.kafka.consumer.cache.maxCapacity 来增大，也可以通过spark.streaming.kafka.consumer.cache.enabled 设置成false 关闭缓存机制。

7) 关于消费者offset

1).如果设置了checkpoint ,那么offset 将会存储在checkpoint中。这种有缺点: 第一，当从checkpoint中恢复数据时，有可能造成重复的消费，需要我们写代码来保证数据的输出幂等。第二，当代码逻辑改变时，无法从checkpoint中来恢复offset.

2).依靠kafka 来存储消费者offset,kafka 中有一个特殊的topic 来存储消费者offset。新的消费者api中，会定期自动提交offset。这种情况有可能也不是我们想要的，因为有可能消费者自动提交了offset,但是后期SparkStreaming 没有将接收来的数据及时处理保存。这里也就是为什么会在配置中将enable.auto.commit 设置成false的原因。这种消费模式也称最多消费一次，默认sparkStreaming 拉取到数据之后就可以更新offset,无论是否消费成功。自动提交offset的频率由参数auto.commit.interval.ms 决定，默认5s。如果我们能保证完全处理完业务之后，可以后期异步的手动提交消费者offset。但是这种将offset存储在kafka中由参数offsets.retention.minutes=1440控制是否过期删除，默认是保存一天，如果停机没有消费达到时长，存储在kafka中的消费者组会被清空，offset也就被清除了。

3).自己存储offset,这样在处理逻辑时，保证数据处理的事务，如果处理数据失败，就不保存offset，处理数据成功则保存offset.这样可以做到精准的处理一次处理数据。

2. SparkStreaming2.3 + Kafka0.11代码

1.	val conf = new SparkConf()
2.	conf.setMaster("local")
3.	conf.setAppName("SparkStreamingOnKafkaDirect")
4.	val ssc = new StreamingContext(conf,Durations.seconds(5))
5.	//设置日志级别
6.	//ssc.sparkContext.setLogLevel("ERROR")
7.	val kafkaParams = Map[String, Object](
8.	    "bootstrap.servers" -> "mynode1:9092,mynode2:9092,mynode3:9092",
9.	    "key.deserializer" -> classOf[StringDeserializer],
10.	    "value.deserializer" -> classOf[StringDeserializer],
11.	    "group.id" -> "MyGroupId",//
12.	    /**
13.	     *
14.	     * earliest ：当各分区下有已提交的offset时，从提交的offset开始消费；无    提交的offset时，从头开始
15.	     * latest：自动重置偏移量为最大偏移量【默认】*
16.	     * none:没有找到以前的offset,抛出异常
17.	     */
18.	    "auto.offset.reset" -> "earliest",
19.	    /**
20.	     * 当设置 enable.auto.commit为false时，不会自动向kafka中保存消费者  offset.需要异步的处理完数据之后手动提交
21.	     */
22.	    "enable.auto.commit" -> (false: java.lang.Boolean)//默认是true
23.	)
24.	
25.	val topics = Array[String]("topic1104")
26.	val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
27.	  ssc,
28.	  PreferConsistent,//消费策略
29.	  Subscribe[String, String](topics, kafkaParams)
30.	)
31.	
32.	val transStrem: DStream[String] = stream.map((record:ConsumerRecord[String, String]) => {
33.	  val key_value = (record.key, record.value)
34.	  println("receive message key = "+key_value._1)
35.	  println("receive message value = "+key_value._2)
36.	  key_value._2
37.	})
38.	val wordsDS: DStream[String] = transStrem.flatMap(line=>{line.split("\t")})
39.	  val result: DStream[(String, Int)] = wordsDS.map((_,1)).reduceByKey(_+_)
40.	  result.print()
41.	
42.	  /**
43.	   * 以上业务处理完成之后，异步的提交消费者offset,这里将 enable.auto.commit  设置成false,就是使用kafka 自己来管理消费者offset
44.	   * 注意这里，获取 offsetRanges: Array[OffsetRange] 每一批次topic 中的 offset时，必须从 源头读取过来的 stream中获取，不能从经过stream转换之后的DStream中获取。
45.	   */
46.	stream.foreachRDD { rdd =>
47.	  val offsetRanges: Array[OffsetRange] =   rdd.asInstanceOf[HasOffsetRanges].offsetRanges
48.	  // some time later, after outputs have completed
49.	  for(or <- offsetRanges){
50.	    println(s"current topic = ${or.topic},partition =  ${or.partition},fromoffset = ${or.fromOffset},untiloffset=${or.untilOffset}")
51.	    }
52.	    stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
53.	  }
54.	ssc.start()
55.	ssc.awaitTermination()
56.	ssc.stop()

3. 手动维护消费者offset

4. 参数配置

Kafka参数配置：

bootstrap.servers 指定Kafka 集群
key.deserializer 指定key的序列化格式
value.deserializer 指定value的序列化格式
group.id 指定消费者组
auto.offset.reset 消费者连接Kafka 消费数据的位置
enable.auto.commit 自动向Kafka提交offset
heartbeat.interval.ms kafka集群与consumer 的心跳【3s】
session.timeout.ms session 会话超时时间【30s】
group.min.session.timeout.ms【6s】
auto.commit.interval.ms 自动提交offset间隔时间

SparkStreaming参数配置：

spark.streaming.blockInterval 200ms
spark.streaming.receiver.maxRate
spark.streaming.receiver.writeAheadLog.enable
spark.streaming.stopGracefullyOnShutdown
spark.streaming.kafka.maxRatePerPartition
spark.streaming.backpressure.enabled
spark.streaming.backpressure.initialRate

本节作业

Kafka 架构原理？
Kafka 命令使用
SparkStreaming + Kafka 整合方式？
SparkStreaming + Kafka 代码？
SparkStreaming + Kafka 配置？

标签：02,25,day36,kafka,topic,--,offset,Kafka,数据
来源： https://blog.csdn.net/qq_44745905/article/details/114034929