Kafka01

1.发布和订阅
2.process  ： kafka自身的stream processing   kafkastreams
3.store ：  Store streams of data safely in a distributed, replicated, fault-tolerant cluster.

kafka如何保证高容错 、保证数据不丢失、？
相当于HDFS的副本机制 eg：
一个文件 打成 多个block块 一个block块 多少个副本

1.发布和订阅
	对应的经典架构是什么呢：
	Flume(生产者)-->Kafka-->SS SSS(消费者)

CDH版本的Kafka ：Kafka是需要自定义部署的   

2.1.1   |	Apache Kafka	| 0.10.0.0+kafka2.1.1+21  
代表什么意思？
0.10.0.0   =》 Apache Kafka的版本
2.1.1  =》  CDH版本 使用Apache Kafka 0.10.0.0 源码编译打补丁 后的 CDH版Kafka  的版本号
21   =》 打了几次补丁的意思

Kafka的版本的选择 要基于 Spark
http://spark.apache.org/docs/latest/streaming-kafka-integration.html

这里我下载最新版本  
el7 和 el6 有什么区别？
el代表 centos的 
 7 -》 7.x  
 6--》6.x  

解压之后 lib里
kafka_2.11-2.2.1-kafka-4.1.0.jar
2.11 scala 版本
2.2.1 apache kafka版本
4.1.0 cdk版本

kafka：
metadata-->zk 
log-->当前机器的Linux磁盘上  (Kakfa存的数据)


部署;
https://blog.csdn.net/jeffleo/article/details/75736474

broker.id=0
host.name=hadoop101
port=9092
log.dirs=/home/double_happy/tmp/kafka-logs00
zookeeper.connect=hadoop101:2181/kafka

broker.id=1
host.name=hadoop101
port=9093
log.dirs=/home/double_happy/tmp/kafka-logs01
zookeeper.connect=hadoop101:2181/kafka

broker.id=2
host.name=hadoop101
port=9094
log.dirs=/home/double_happy/tmp/kafka-logs02
zookeeper.connect=hadoop101:2181/kafka


启动：
bin/kafka-server-start.sh -daemon config/server.properties
bin/kafka-server-start.sh -daemon config/server-1.properties
bin/kafka-server-start.sh -daemon config/server-2.properties

[zk: localhost:2181(CONNECTED) 4] ls /
[zookeeper, kafka]
[zk: localhost:2181(CONNECTED) 5]

[double_happy@hadoop101 kafka]$ jps
17088 QuorumPeerMain
20992 Kafka
4289 NodeManager
21810 Jps
4019 SecondaryNameNode
21591 Kafka
14999 AzkabanSingleServer
20392 Kafka
6633 HistoryServer
3721 NameNode
4186 ResourceManager
3853 DataNode

4.几个概念 
producer: 生产者 Flume Maxwell
consumer：消费者 SS/SSS/Flink/Flume
broker： 消息处理节点  kafka的启动进程

topic: 主题 
	oms订单系统mysql.oms -->maxwell-->kafka  topic: oms  /oms文件夹下
	wms仓库系统mysql.wms -->maxwell-->kafka  topic: wms  /wms文件夹下

	app log---flume-->kafka topic： applog
	systemlog---flume-->kafka topic： systemlog

partition：分区 topic物理上的分组 (对应的Linux文件夹)，
           一个topic 可以分>=1个p，每个p是一个有序的队列：数据落地到磁盘 是0拷贝+按顺序追加写
	  这就是kafka高效的数据中间件的原因
      按顺序追加写 ： 磁盘的那个指针按一个方向追加和读写的 
           oms_0 oms_1 oms_2

这个parition ：
相当于HDFS的副本机制 eg：
一个文件 打成 多个block块 一个block块 多少个副本 

kafka里的分区就相当于 HDFS里的分块的概念 

replication：副本数   和HDFS的副本数 一样的

常用命令：

[double_happy@hadoop101 bin]$ ./kafka-topics.sh 
Create, delete, describe, or change a topic.
Option                                   Description                            
------                                   -----------                            
--alter                                  Alter the number of partitions,        
                                           replica assignment, and/or           
                                           configuration for the topic.         
--bootstrap-server <String: server to    REQUIRED: The Kafka server to connect  
  connect to>                              to. In case of providing this, a     
                                           direct Zookeeper connection won't be 
                                           required.                            
--command-config <String: command        Property file containing configs to be 
  config property file>                    passed to Admin Client. This is used 
                                           only with --bootstrap-server option  
                                           for describing and altering broker   
                                           configs.                             
--config <String: name=value>            A topic configuration override for the 
                                           topic being created or altered.The   
                                           following is a list of valid         
                                           configurations:                      
                                                cleanup.policy                        
                                                compression.type                      
                                                delete.retention.ms                   
                                                file.delete.delay.ms                  
                                                flush.messages                        
                                                flush.ms                              
                                                follower.replication.throttled.       
                                           replicas                             
                                                index.interval.bytes                  
                                                leader.replication.throttled.replicas 
                                                max.message.bytes                     
                                                message.downconversion.enable         
                                                message.format.version                
                                                message.timestamp.difference.max.ms   
                                                message.timestamp.type                
                                                min.cleanable.dirty.ratio             
                                                min.compaction.lag.ms                 
                                                min.insync.replicas                   
                                                preallocate                           
                                                retention.bytes                       
                                                retention.ms                          
                                                segment.bytes                         
                                                segment.index.bytes                   
                                                segment.jitter.ms                     
                                                segment.ms                            
                                                unclean.leader.election.enable        
                                         See the Kafka documentation for full   
                                           details on the topic configs.It is   
                                           supported only in combination with --
                                           create if --bootstrap-server option  
                                           is used.                             
--create                                 Create a new topic.                    
--delete                                 Delete a topic                         
--delete-config <String: name>           A topic configuration override to be   
                                           removed for an existing topic (see   
                                           the list of configurations under the 
                                           --config option). Not supported with 
                                           the --bootstrap-server option.       
--describe                               List details for the given topics.     
--disable-rack-aware                     Disable rack aware replica assignment  
--exclude-internal                       exclude internal topics when running   
                                           list or describe command. The        
                                           internal topics will be listed by    
                                           default                              
--force                                  Suppress console prompts               
--help                                   Print usage information.               
--if-exists                              if set when altering or deleting or    
                                           describing topics, the action will   
                                           only execute if the topic exists.    
                                           Not supported with the --bootstrap-  
                                           server option.                       
--if-not-exists                          if set when creating topics, the       
                                           action will only execute if the      
                                           topic does not already exist. Not    
                                           supported with the --bootstrap-      
                                           server option.                       
--list                                   List all available topics.             
--partitions <Integer: # of partitions>  The number of partitions for the topic 
                                           being created or altered (WARNING:   
                                           If partitions are increased for a    
                                           topic that has a key, the partition  
                                           logic or ordering of the messages    
                                           will be affected                     
--replica-assignment <String:            A list of manual partition-to-broker   
  broker_id_for_part1_replica1 :           assignments for the topic being      
  broker_id_for_part1_replica2 ,           created or altered.                  
  broker_id_for_part2_replica1 :                                                
  broker_id_for_part2_replica2 , ...>                                           
--replication-factor <Integer:           The replication factor for each        
  replication factor>                      partition in the topic being created.
--topic <String: topic>                  The topic to create, alter, describe   
                                           or delete. It also accepts a regular 
                                           expression, except for --create      
                                           option. Put topic name in double     
                                           quotes and use the '\' prefix to     
                                           escape regular expression symbols; e.
                                           g. "test\.topic".                    
--topics-with-overrides                  if set when describing topics, only    
                                           show topics that have overridden     
                                           configs                              
--unavailable-partitions                 if set when describing topics, only    
                                           show partitions whose leader is not  
                                           available                            
--under-replicated-partitions            if set when describing topics, only    
                                           show under replicated partitions     
--zookeeper <String: hosts>              DEPRECATED, The connection string for  
                                           the zookeeper connection in the form 
                                           host:port. Multiple hosts can be     
                                           given to allow fail-over.            
[double_happy@hadoop101 bin]$

1.创建topic

[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --create \
> --zookeeper hadoop101:2181/kafka \
> --replication-factor 3 \
> --partitions 3 \
> --topic test
Created topic test.
[double_happy@hadoop101 kafka]$

注意：
1.会创建3个文件夹 
test-0
test-1
test-2

test : topic 的名字   012 是partition的 下标从0开始 

--replication-factor 3   三个副本指的是  我们有三台机器 副本数要 <= 3     你三台机器 不能创建多余机器数的副本数

2.查看我们的topic  ： 查看我当前的kafka集群有多少个Topic
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --list \
> --zookeeper hadoop101:2181/kafka 
test
[double_happy@hadoop101 kafka]$

3.查看 topic 明细   ***
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --describe \
> --zookeeper hadoop101:2181/kafka \
> --topic test
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,1,0 Isr: 2,1,0
        Topic: test     Partition: 2    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
[double_happy@hadoop101 kafka]$ 

解释：因为我是单节点部署三台kafka 这样说不好理解 我就把他变成三台机器部署三台kfaka 

1.机器
broker 0  -> Hadoop101
broker 1 -> Hadoop102
broker 2 -> Hadoop103

2.明细
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 1   Hadoop102      Replicas: 1,0,2 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 2   Hadoop103   Replicas: 2,1,0 Isr: 2,1,0
        Topic: test     Partition: 2    Leader: 0   Hadoop101    Replicas: 0,2,1 Isr: 0,2,1

3.解释
Leader、Replicas、Isr：里面的数字 代表  broker id
Partition：数字 代表 paritition的下标
Leader ：该分区 负责对外的 读写的节点 

3.1    按 分区0 进行解释 
对于test 这个主题来说，我们的数据完整性 是由三个分区 来组成的  012 partition

就是说 我0分区 由分本数 三个副本 三个副本是分散在不同的机器上面的
但是哪个机器进行对外读和写呢 ？是由Leader 的值来表现的 

那么：
0 分区的数据 是由 哪台机器 来进行对外读写 提供的(Leader 1) 说明是 broker 1 号机器 Hadoop102

那么 ：
 Replicas: 1,0,2   表示 复制分区的节点的列表 
 0,2 =》表示 Hadoop103、Hadoop101    这两台机器对parittion 0 进行数据的复制同步

Isr：in-sync 表示正在同步的集群列表

那么：
isr: 1,0,2  
	1表示对外的 leader
	假设分区0 对应的leader 1 机器挂了 ，那么 0 将会作为leader 进行对外提供服务



关键点：
1.分区  topic的数据完整性
2.谁负责对外读写
3.谁负责分区数据同步

测试：isr

[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh --describe --zookeeper hadoop101:2181/kafka --topic test
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,1,0 Isr: 2,1,0
        Topic: test     Partition: 2    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
[double_happy@hadoop101 kafka]$ jps
17088 QuorumPeerMain
4289 NodeManager
4019 SecondaryNameNode
24260 Kafka
23653 Kafka
14999 AzkabanSingleServer
6633 HistoryServer
3721 NameNode
4186 ResourceManager
3853 DataNode
24862 Kafka
31278 Jps
[double_happy@hadoop101 kafka]$ kill -9 24260
[double_happy@hadoop101 kafka]$ 
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh --describe --zookeeper hadoop101:2181/kafka --topic test
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 1,0,2 Isr: 0,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,1,0 Isr: 2,0
        Topic: test     Partition: 2    Leader: 0       Replicas: 0,2,1 Isr: 0,2
[double_happy@hadoop101 kafka]$

很巧 瞎kill一个真的把 1 broker kill掉了 

注意;
	看paritition 0  是不是 1被干掉之后 0上位了 当成了leader
2.你kill -9  是不是太暴力了 
查看

[double_happy@hadoop101 bin]$ cat kafka-server-stop.sh
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
SIGNAL=${SIGNAL:-TERM}
PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')

if [ -z "$PIDS" ]; then
  echo "No kafka server to stop"
  exit 1
else
  kill -s $SIGNAL $PIDS
fi
[double_happy@hadoop101 bin]$ 


官方使用 kill -s  接受一个信号    有兴趣了解一下 但是官方也用了 kill

重启 broker 1 ：
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh --describe --zookeeper hadoop101:2181/kafka --topic test
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 1,0,2 Isr: 0,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,1,0 Isr: 2,0
        Topic: test     Partition: 2    Leader: 0       Replicas: 0,2,1 Isr: 0,2
[double_happy@hadoop101 kafka]$ bin/kafka-server-start.sh -daemon config/server-1.properties
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh --describe --zookeeper hadoop101:2181/kafka --topic test
 
Topic:test      PartitionCount:3        ReplicationFactor:3     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 1,0,2 Isr: 0,2,1
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,1,0 Isr: 2,0,1
        Topic: test     Partition: 2    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
[double_happy@hadoop101 kafka]$

4.删除 Topic
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --delete \
> --zookeeper hadoop101:2181/kafka \
> --topic test
Topic test is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
[double_happy@hadoop101 kafka]$ 


注意：
Topic test is marked for deletion.   已经标识了删除  那么我们还能不能查看到呢？

[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh --delete --zookeeper hadoop101:2181/kafka --topic test
Error while executing topic command : Topics in [] does not exist
[2019-10-30 11:06:04,714] ERROR java.lang.IllegalArgumentException: Topics in [] does not exist
        at kafka.admin.TopicCommand$.kafka$admin$TopicCommand$$ensureTopicExists(TopicCommand.scala:416)
        at kafka.admin.TopicCommand$ZookeeperTopicService.deleteTopic(TopicCommand.scala:377)
        at kafka.admin.TopicCommand$.main(TopicCommand.scala:68)
        at kafka.admin.TopicCommand.main(TopicCommand.scala)
 (kafka.admin.TopicCommand$)
[double_happy@hadoop101 kafka]$

查看不到了 说明Topic 已经不存在了 

去机器上 看一下目录

正常应该由 delete 标识的  等重启Kafak之后 会彻底删除 delete 标识的文件夹  

我这里没有演示出来  

Note: This will have no impact if delete.topic.enable is not set to true.

这个提示是 表示;
delete.topic.enable  参数没有设值为 true  那么 删除不会立马起作用 
(双重否定 表示 肯定   是true 就会立马删除 )

去官网看看：
delete.topic.enable 默认 就是 true  
所以 提示这句话 就是 放屁 
但是也要注意 生产上别随便删除 Topic 删了可能会导致 kafka起不来 真的 别没事找事做

我想总结的是这个 删除topic 可能会导致kafka故障？
删除整理:
1.⽣产上命名不要有标点符号的字符，就英⽂ 可以带数字 默认⼩写
2.假如已经在运⾏的kafka 只有1个topic，你可以删除，0⻛险
3.假如已经在运⾏的kafka 只有多个topic，忍，⻛险可能存在  (你把自己的topic 删完 kafka坏了 别人的topic也没法用了 )
4.千万不要⼿贱或者强迫症，看topic名称不舒服，去删除topic，删除需谨慎！
5.删除不可逆，细⼼操作删除命令！

删除可能遇到的问题？当你真的要删除Topic的时候 
       如果可以重启kafka 就重启kafka 不能重启看下面
假如删除不⼲净:
1.linux磁盘⽂件夹  手工删除 
2.zk的/kafka的
ls /kafka/brokers/topics
ls /kafka/config/topics
默认delete.topic.enable=true，执⾏删除命令后，⽆需关⼼。(因为会立马删除掉 )

看一下刚刚删除完的 topic 在zk的元数据

[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper, kafka]
[zk: localhost:2181(CONNECTED) 1] ls /kafka/brokers/topics
[]
[zk: localhost:2181(CONNECTED) 2] ls /kafka/config/topics
[]
[zk: localhost:2181(CONNECTED) 3] 

没有 说明删的 很干净

5..修改topic

[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --create \
> --zookeeper hadoop101:2181/kafka \
> --replication-factor 1 \
> --partitions 1 \
> --topic g7
Created topic g7.
[double_happy@hadoop101 kafka]$ 


（1）修改分区数量 
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --alter \
> --zookeeper hadoop101:2181/kafka \
> --topic g7 --partitions 3
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!
[double_happy@hadoop101 kafka]$

没有 g7-1 g7-2
说明分区的副本只有一份   其他分区落不进来

副本数 一般3分 
broker 8台机器 qbs很高的 生产上 是这么用的   3，4万的qbs

查看明细：
[double_happy@hadoop101 kafka]$ bin/kafka-topics.sh \
> --describe \
> --zookeeper hadoop101:2181/kafka \
> --topic g7 
Topic:g7        PartitionCount:3     ReplicationFactor:1      Configs:
        Topic: g7       Partition: 0 Leader: 2        Replicas: 2     Isr: 2
        Topic: g7       Partition: 1 Leader: 0        Replicas: 0     Isr: 0
        Topic: g7       Partition: 2 Leader: 1        Replicas: 1     Isr: 1
[double_happy@hadoop101 kafka]$ 

说明副本只有一个  isr 只有一个 挂了 就丢数据了

6.自动迁移数据到 新的节点 ：

假如有一天 你的kafka需要添加机器 那么你之前的kafka数据 要进行迁移

1.  kafka-reassign-partitions.sh    使用这个脚本 
重新分配我们的partition  这种情况下主要是针对 
kafka的集群规模 数据量 扛不住了
需要新加机器来平衡我的数据量 

就三步：
1.生成一个json的文件 
bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --topics-to-move-json-file topics-to-move.json --broker-list "5,6" --generate

就这个命令 会生成 json文件 


--broker-list "5,6"     表示 1234 下架 只用56   如果你用1356  那么这个参数这么写呢 ？
                    --broker-list "1,3,5,6"   明白了吗？

2.json文件就是一个标准 后面根据它 进行重新分配    (就是根据json 去执行)
bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --execute

3. 验证第二步  (进行校验)
bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file expand-cluster-reassignment.json --verify

工作时候 用的很少 一般都是事先规划好的 机器

7.console案例   很简单 

生产者：
[double_happy@hadoop101 kafka]$ bin/kafka-console-producer.sh \
> --broker-list hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic g7
>


消费者：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic g7 \
> --from-beginning


这就模拟了 生产者 发送数据到 broker  然后消费者进行消费

生产者：
[double_happy@hadoop101 kafka]$ bin/kafka-console-producer.sh \
> --broker-list hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic g7
>doubleHappy
>1
>2
>3
>4
>5
>6
>7
>8
>9
>

消费者：
[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh \
> --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 \
> --topic g7 \
> --from-beginning
doubleHappy
1
2
3
4
5
6
7
8
9

那么我把消费者 打断 重新 起来 ：

--from-beginning  表示 从头开始消费    表示消息从最初始的位置开始消费 

[double_happy@hadoop101 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop101:9092,hadoop101:9093,hadoop101:9094 --topic g7 --from-beginning
1
4
7
2
5
8
doubleHappy
3
6
9




问题：
是不是乱序了   指的是全局乱序的   因为kafka 单分区 数据是有序的 (就是 每个分区内的数据 在各自当前分区是有序的 )

数据在生产的时候 发送到kafka的时候
 数据是发往不同的分区里 

不同分区里的数据应该是这样的：
1 4 7     |    2 5 8  |  doubleHappy  3 6 9

**数据在每个 分区内部是有序的    

那么 如何保证数据 全局有序？
    自己思考思考 我下篇文章写