Kafka搭建及使用

Kafka搭建

1、上传解压修改环境变量

# 解压
tar -zxvf kafka_2.11-1.0.0.tgz -C /usr/local/soft
mv kafka_2.11-1.0.0 kafka-1.0.0

tar -xvf 是一个在Unix和类Unix操作系统（如Linux和macOS）中用于解压缩或解包.tar文件的命令。
tar -zxvf 是一个在Unix和类Unix操作系统（如Linux和macOS）中用于解压缩.tar.gz或.tgz文件的命令。
tar：是“tape archive”的缩写，用于文件的归档和压缩。
-z：这个选项告诉tar命令接下来要处理的文件是通过gzip进行压缩的。因此，tar会在解压归档文件之前先解压gzip压缩。
-x：这个选项表示要进行的是解包（或解压）操作，即从归档文件中提取文件。
-v：这个选项代表“verbose”，意味着命令在执行时会显示详细的操作信息，比如正在解压的文件名。
-f：这个选项后面跟着要操作的文件名，告诉tar命令接下来要处理的文件名。注意，-f选项和文件名之间不应该有空格。


# 配置环境变量
vim /etc/profile

export KAFKA_HOME=/usr/local/soft/kafka-1.0.0
export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

2、修改配置文件

vim config/server.properties

# 2181：zookeeper端口
broker.id=0 每一个节点broker.id 要不一样
zookeeper.connect=master:2181,node1:2181,node2:2181/kafka
log.dirs=/usr/local/soft/kafka-1.0.0/data   数据存放的位置

3、将kafka文件同步到node1,node2

# 同步kafka文件   -r 选项表示递归复制，这对于复制目录及其所有子目录和文件是必要的。
scp -r kafka-1.0.0/ node1:`pwd`
scp -r kafka_2.11-1.0.0/ node2:`pwd`

# 修改node1、node2中的/etc/profile，增加Kafka环境变量
export KAFKA_HOME=/usr/local/soft/kafka-1.0.0
export PATH=$PATH:$KAFKA_HOME/bin

#  在ndoe1和node2中执行source
source /etc/profile

4、修改node1和node2中的broker.id

vim config/server.properties

# node1
broker.id=1
# node2
broker.id=2

5、启动kafka

# 1、需要启动zookeeper,  kafka使用zk保存元数据
# 需要在每个节点中执行启动的命令
zkServer.sh start
# 查看启动的状体
zkServer.sh status

#若不清楚指令用法，例如：可以使用kafka-console-producer.sh help来查找对应参数用法

# 2、启动kafka，每个节点中都要启动（去中心化的架构）
# -daemon后台启动
kafka-server-start.sh -daemon /usr/local/soft/kafka-1.0.0/config/server.properties

#注：由于是去中心化的，所以指定一台即可(master:9092,node1:9092,node2:909),但是怕它会宕机，所以保险起见都写上
# 测试是否成功
#生产者 --topic shujia:指定topic名称；9092：kafka端口
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic shujia

# 消费者
 --from-beginning   从头消费，如果不在执行消费的新的数据
kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic shujia


#进入zookeeper的命令行界面，查看zookeeper上kafka的元数据信息
zkServer.sh
#可以help查询操作帮助
help

#错误？？？
ZooKeeper JMX enabled by default
Using config: /usr/local/soft/zookeeper-3.5.7/bin/../conf/zoo.cfg
Usage: ./zkServer.sh [--config <conf-dir>] {start|start-foreground|stop|restart|status|print-cmd}

Kafka使用

1、创建topic

在生产和消费数据时，如果topic不存在会自动创建一个分区为1，副本为1的topic

--replication-factor  ---每一个分区的副本数量, 同一个分区的副本不能放在同一个节点，副本的数量不能大于kafak集群节点的数量
--partition   --分区数，  根据数据量设置
--zookeeper zk的地址，将topic的元数据保存在zookeeper中；/kafka:指定创建Topic的元数据在zookeeper中的路径

kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181/kafka --replication-factor 2 --partitions 3 --topic bigdata

进入zookeeper的命令行，查看里面所存储的元数据信息

启动：zkServer.sh

进行命令行界面：zkCli.sh

2、查看topic描述信息

kafka-topics.sh --describe --zookeeper master:2181,node1:2181,node2:2181/kafka --topic bigdata

• Topic: 主题名称，这里是bigdata。
• Partition: 分区编号，用于标识主题中的具体分区，从0开始编号。（除partition：0中的数字代表分区，后面的数字均代表节点）
• Leader: 领导副本（Leader Replica）的编号。每个分区都有一个领导副本，所有的生产者（Producer）和消费者（Consumer）都通过这个领导副本来读写数据。如果领导副本不可用，Kafka会自动从ISR中选择一个新的领导副本。
• Replicas: 副本列表，列出了所有副本的编号。这些副本分布在不同的Kafka broker上，以提供数据的冗余和容错。
• ISR: 同步副本（In-Sync Replicas）列表，是Replicas的一个子集。ISR中的副本与领导副本保持同步，即它们已经复制了领导副本上的所有消息。只有当副本与领导副本保持同步时，它才会被包含在ISR中。ISR的存在是为了确保在发生故障转移时，新的领导副本能够拥有最新的数据。

3、获取所有topic

__consumer_offsetsL kafka用于保存消费便宜量的topic

kafka-topics.sh --list --zookeeper master:2181,node1:2181,node2:2181/kafka

4、创建控制台生产者

kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic bigdata

5、创建控制台消费者

 --from-beginning   从头消费 如果不在执行消费的新的数据
kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node3:9092 --from-beginning --topic bigdata

kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node3:9092 --from-beginning --topic students1000
 

6、kafka数据保存的方式

# 1、保存的文件
/usr/local/soft/kafka_2.11-1.0.0/data

# 2，每一个分区每一个副本对应一个目录

# 3、每一个分区目录中可以有多个文件， 文件时滚动生成的
00000000000000000000.log
00000000000000000001.log
00000000000000000002.log

# 4、滚动生成文件的策略
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

# 5、文件删除的策略，默认时7天，以文件为单位删除
#可以使用 --delete实现删除的标记，再次查询时则查不到该topic，但是数据依旧存在；可以通过配置实现真正的删除，但是我们无需配置
log.retention.hours=168

Flink整合kafka

1、idea中整合

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${flink.version}</version>
</dependency>

2、集群中整合

# 将flink-sql-connector-kafka-{flink.version}.jar包上传到Flink的lib目录下
cd /usr/local/soft/flink-{flink.version}/lib