Kafka 相关

Kafka 相关 Kafka

定义规则，检测是否满足规则，并且做出这个规则里所做的动作。【规则的计算和匹配性，一切皆动态规则】

使用场景：实时推荐、实时风控、实时精准广告推送。

[实时推荐] 冷用户访问 --> 给优惠券；

[实时推荐] 只看不买 --> 促单；

[实时推荐] 生成订单后未支付，给发短信。

[实时风控] 某IP近1小时内注册账号超过10个； 某账号群体近1h内购买优惠券商品超过100件...

1. kafka 的工作原理：

source :  Kafka学习之路 （一）Kafka的简介 - 扎心了，老铁 - 博客园 (cnblogs.com)

source ：Kafka学习之路 （二）Kafka的架构 - 扎心了，老铁 - 博客园 (cnblogs.com)

source ：Kafka学习之路 （三）Kafka的高可用 - 扎心了，老铁 - 博客园 (cnblogs.com)

source :  Kafka学习之路 （四）Kafka的安装 - 扎心了，老铁 - 博客园 (cnblogs.com)

source :  Kafka学习之路 （五）Kafka在zookeeper中的存储 - 扎心了，老铁 - 博客园 (cnblogs.com)

这个人的东西挺好的，很多东西都很值得看看。比如说spark系列...

1.1.1 消息系统的两种模式

通常一个消息系统， 主要有两种模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

点对点传递模式：生产者将消息持久化队列后，只有一个consumer 可以消费到，而且只可以消费一次，消费之后该数据从队列queue中删除。保证了即使有多个消费者同时消费数据也能保证数据处理的顺序。

​        

发布-订阅模式：在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。

​

1.1.2kafka 中术语解释

broker：Kafka 集群包含一个或多个服务器，服务器节点称为broker。

broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。

如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。

如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

topic ：{类似于数据库的表名}。每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

partition ： topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，不同partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

Producer ：生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

customer ：消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

Consumer Group ：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

Leader ：每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

Follower ：Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。

Topics和Partition

Producer消息路由

Consumer Group

Push vs. Pull

        作为一个消息系统，Kafka遵循了传统的方式，选择由Producer向broker push消息并由Consumer从broker pull消息。

        一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume，采用push模式。事实上，push模式和pull模式各有优劣。

        push模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。push模式的目标是尽可能以最快速度传递消息，但是这样很容易造成Consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据Consumer的消费能力以适当的速率消费消息。

        对于Kafka而言，pull模式更合适。pull模式可简化broker的设计，Consumer可自主控制消费消息的速率，同时Consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

Kafka delivery guarantee

1.1.3 kafka 拓扑结构图

​

        如上图所示，一个典型的Kafka集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

2. kafka 的使用指南:

source : Kafka核心API——Connect API - 阅读清单 - 云+社区 - 腾讯云

source : Kafka Connect介绍以及在分布式部署模式下启动流程分析 - 墨天轮

2.1 下载并安装

(2-1) 启动服务

运行kafka需要使用Zookeeper，所以需要先启动一个Zookeeper服务器，如果没有Zookeeper，可以使用kafka自带打包和配置好的Zookeeper.其中daemon 是在后台进行该进程.

# bin/zookeeper-server-start.sh config/zookeeper.properties

(2-2) 然后启动kafka服务

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

(3-1) 新建一个topic

创建一个名为“test”的Topic，只有一个分区和一个备份：

# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

创建好之后，可以通过以下命令查看已创建的topic信息：(除手工创建topic外，也可以配置broker，当发布一个不存在的topic时自动创建topic。)

# bin/kafka-topics.sh --list --zookeeper localhost:2181test

(3-2) 查看topic

bin/kafka-topics.sh --zookeeper localhost:2181 --list

（3-3）查看topic 中的内容

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicName --from-beginning

(3-4) 删除topic

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic 要删掉的topicname

(4) kafka连接器 Kafka Connect的工作模式分为两种，分别是standalone模式和distributed模式。

connect-standalone.sh 用于测试开发

connect-distributed.sh 用于正式环境线上

(4-1) standalone模式 

(4-2) 开启distributed模式

bin/connect-distributed.sh -daemon config/connect-distributed.properties

(5) 杀死进程

kill -9 进程号

(6) 查看进程

jps

(7-1) 发送消息  Kafka提供了一个命令行工具，可以从输入文件或者命令行中读取消息并发送给Kafka集群，每一行是一条消息。运行producer,然后在控制台输入几条消息到服务器

# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

(7-2) 消费消息   Kafka也提供了一个消费消息的命令行工具

# bin/kafka-console-consumer.sh --bootstrap-server 172.16.2.120:9092 --topic topic的name --from-beginning

(8) #查看kafka的安装目录  

kafka 基本指令:

source : Apache Kafka 入门 - Kafka命令详细介绍_偶尔记一下 - mybatis.io-CSDN博客

kafka 参数属性source : Kafka常用命令之kafka-console-consumer.sh_Ernest-CSDN博客_kafka-console-consumer.sh

topic 配置:

Topic配置 · kafka-tutorial-cn

3. Kafka Connect简介

        Kafka 0.9+增加了一个新的特性Kafka Connect,可以更方便的创建和管理数据流管道。它为Kafka和其它系统创建规模可扩展的、可信赖的流数据提供了一个简单的模型，通过connectors可以将大数据从其它系统导入到Kafka中，也可以从Kafka中导出到其它系统。Kafka Connect可以将完整的数据库注入到Kafka的Topic中，或者将服务器的系统监控指标注入到Kafka，然后像正常的Kafka流处理机制一样进行数据流处理。而导出工作则是将数据从Kafka Topic中导出到其它数据存储系统、查询系统或者离线分析系统等，比如数据库、Elastic Search、Apache Ignite等。

Kafka Connect特性包括：

Kafka connector通用框架,提供统一的集成API同时支持分布式模式和单机模式REST 接口，用来查看和管理Kafka connectors自动化的offset管理，开发人员不必担心错误处理的影响分布式、可扩展流/批处理集成

​

Connectors 连接器，分为两种 Source（从源数据库拉取数据写入Kafka），Sink（从Kafka消费数据写入目标数据）

连接器其实并不参与实际的数据copy，连接器负责管理Task。连接器中定义了对应Task的类型，对外提供配置选项（用户创建连接器时需要提供对应的配置信息）。并且连接器还可以决定启动多少个Task线程。用户可以通过Rest API 启停连接器，查看连接器状态.

Task

实际进行数据传输的单元，和连接器一样同样分为 Source和Sink

Task的配置和状态存储在Kafka的Topic中，config.storage.topic和status.storage.topic。我们可以随时启动，停止任务，以提供d性、可扩展的数据管道

Worker

刚刚我们讲的Connectors 和Task 属于逻辑单元，而Worker 是实际运行逻辑单元的进程，Worker 分为两种模式，单机模式和分布式模式

单机模式：比较简单，但是功能也受限，只有一些特殊的场景会使用到，例如收集主机的日志，通常来说更多的是使用分布式模式

分布式模式：为Kafka Connect提供了可扩展和故障转移。相同group.id的Worker，会自动组成集群。当新增Worker，或者有Worker挂掉时，集群会自动协调分配所有的Connector 和 Task（这个过程称为Rebalance）

4. Spark Streaming获取kafka数据的两种方式

source : Spark Streaming获取kafka数据的两种方式_B11050101的博客-CSDN博客_kafka拉取数据的两种方式

Spark Streaming读取Kafka数据的两种方式 - Boblim - 博客园

4. kafka Question：

(1) 获取kafka 数据,怎么获取到历史的数据了呢?  因为你的偏移量设置为 earliest 了呀，这就是要从头开始读取了呢。

val dataStreamReader = spark.readStream.format("kafka").option("startingOffsets", "earliest")