2024.1.13 Kafka六大机制和Structured Streaming

                轮询分发策略: 在Kafka的老版本中存在的一种分发策略，当生产数据的时候，只有value但是没有key的时候，采用轮询。
   优点: 可以保证每个分区拿到的数据基本是一样，因为是一个一个的轮询的分发
   缺点: 如果采用异步发送方式，意味着一批数据发送到broker端，由于是轮询策略，会将这一批数据拆分为多个小的批次，分别再写入到不同的分区里面去，写入进去以后，每个分区都会给予响应，会影响写入效率。

                粘性分发策略: 在Kafka新版本中存在的一种分发策略。当生产数据的时候，只有value但是没有key的时候，采用粘性分发策略
   优点: 在发送数据的时候，首先会随机的选取一个分区，然后尽可能将数据分发到这个分区上面去，也就是尽可能粘着这个分区。该分发方式，在异步发送的操作中，效率比较高。
   缺点: 在数据发送特别快的时候，可能会导致某个分区的数据比其他分区数据多很多，造成大量的数据集中在一个分区上面

二. Kafka消费者的负载均衡机制

Kafka消费者的负载均衡机制
1- 在同一个消费组中，消费者的个数最多不能超过Topic的分区数。如果超过了，就会有一些消费者处于闲置状态，消费不到任何数据。
2- 在同一个消费组中，一个Topic中一个分区的数据，只能被同个消费组中的一个消费者所消费，不能被同个消费组中多个消费者所消费。但是一个消费组内的一个消费者可以消费多个分区的数据。也就是分区和消费者的对应关系，多对一
3-不同的消费组中的消费者，可以对一个Topic的数据同时消费，也就是不同消费组间没有任何关系

三 . 数据不丢失机制

生产者端是如何保证数据不丢失的呢？

答：生产者端将消息发送给到Kafka集群以后，broker要给生产者响应信息。响应原理就是ACK机制

ACK机制当中有3个参数配置值，分别是：0 1 -1(all)
0：生产者生产消息给到Kafka集群，生产者不等待（不接收）broker返回的响应信息
1：生产者生产消息给到Kafka集群，Kafka集群中的分区对应的Leader主副本所在的broker给生产者返回响应信息
-1(all)：生产者生产消息给到Kafka集群，Kafka集群中的分区对应的所有副本给生产者返回响应信息

消息的生产效率排序（由高到低）：0 > 1 > -1
消息的安全级别排序（由高到低）：-1 > 1 > 0

在实际工作中如何选择ACK参数配置？
答：根据数据的重要程度进行选择。如果数据重要，优先保证数据的安全性，再考虑生产效率；如果数据不重要，优先考虑生产效率，再尽可能提升安全级别。

Broker端如何保证数据不丢失

Broker端通过多副本机制确保数据不丢失。同时需要生产者端将acks设置为-1

消费端如何保证数据不丢失

消费者消费消息的步骤：
1- 消费者首先连接到Kafka集群中，进行消息的消费

2- Kafka集群接收到Consumer消费者的消费请求以后，首先会根据group id（消费组名称），查找上次消费消息对应的offset（偏移量）

3- 如果没有查找到offset，消费者默认从Topic最新的地方开始消费

4- 如果有查找到offset，会从上次消费到的offset地方进行继续消费
   4.1- 首先先确定要读取的这个offset偏移量在哪个segment文件当中
   4.2- 查询这个segment文件对应的index文件，根据offset确定这个消息在log文件的什么位置，也就是确定消息的物理偏移量
   4.3- 读取log文件，查询对应范围内的数据即可
   4.4- 获取最终的消息数据

5- 消费者在消费的过程中，底层有个线程会定时的将消费的offset提交给到Kafka集群。Kafka集群会更新对应的offset的值

Kafka中消费者如何对数据仅且只消费一次

1- 将消费者的 enable.auto.commit 属性设置为 false，并手动管理消费者的偏移量。这样可以确保消费者在处理完所有消息后才更新偏移量，避免重复消费数据。也就是将消息的消费、消息业务处理代码、offset提交代码放在同一个事务当中。

2- 使用幂等生产者或事务性生产者来确保消息只被发送一次。这样可以避免重复发送消息，从而避免消费者重复消费数据。

3- 在消息中加入唯一的ID

四 . 启动Kafka eagle命令

cd /export/server/kafka-eagle-bin-1.4.6/kafka-eagle-web-1.4.6/bin

./ke.sh start

结构化流测试linux开启命令

首先: 先下载一个 nc(netcat) 命令. 通过此命令打开一个端口号, 并且可以向这个端口写入数据
yum -y install nc

执行nc命令, 开启端口号, 写入数据:
nc -lk 55555

注意: 要先启动nc，再启动我们的程序

查看端口号是否被使用命令：
netstat -nlp | grep 要查询的端口

数据积压问题处理

出现积压的原因:

因为数据写入目的容器失败,从而导致消费失败
因为网络延迟消息消费失败
消费逻辑过于复杂, 导致消费过慢,出现积压问题

解决方案:

对于第一种, 我们常规解决方案, 处理目的容器,保证目的容器是一直可用状态
对于第二种, 如果之前一直没问题, 只是某一天出现, 可以调整消费的超时时间。并且同时解决网络延迟问题
对于第三种, 一般解决方案,调整消费代码, 消费更快即可, 利于消费者的负载均衡策略,提升消费者数量

五 . 结构化流

有界: 数据大小固定,有开始和结尾

无界: 源源不断的数据,没有明确的结尾

结构化流是构建在Spark SQL处理引擎之上的一个流式的处理引擎，主要是针对无界数据的处理操作。对于结构化流同样也支持多种语言操作的API：比如 Python Java Scala SQL ….

Spark的核心是RDD。RDD出现主要的目的就是提供更加高效的离线的迭代计算操作，RDD是针对的有界的数据集，但是为了能够兼容实时计算的处理场景，提供微批处理模型，本质上还是批处理，只不过批与批之间的处理间隔时间变短了，让我们感觉是在进行流式的计算操作，目前默认的微批可以达到100毫秒一次

真正的流处理引擎: Flink、Storm(早期流式处理引擎)、Flume(流式数据采集)

数据源 File Source

将目录中写入的文件作为数据流读取，支持的文件格式为：text、csv、json、orc、parquet。。。。

文件数据源特点：
        1- 不能够监听具体的文件，否则会报错误java.lang.IllegalArgumentException: Option ‘basePath’ must be a directory
        2- 可以通过通配符的形式，来监听目录下的文件，符合要求的才会被读取
        3- 如果监听目录中有子目录，那么无法监听到子目录的变化情况

File source只能监听目录，不能监听具体文件

读取代码通用格式:

sparksession.readStream

.format(‘CSV|JSON|TEXT|PARQUET|ORC)

.option(‘参数名1′,’参数值1’)
                    .option(‘参数名2′,’参数值2’)
                    .option(‘参数名N’,’参数值N’)
                    .schema(元数据信息)
                    .load(‘需要监听的目录地址’)