大数据组件使用 总文章

• kafka 生产/消费API、offset管理/原理、kafka命令
• kafka 命令、API
• Kafka 安装、原理、使用
• mapreduce 实时消费 kafka 数据

创建topic
kafka-topics --create --zookeeper node1:2181 --replication-factor 3 --partitions 3 --topic test

展示topic
kafka-topics --list --zookeeper node1:2181
 
描述topic
kafka-topics --describe --zookeeper node1:2181 --topic test

生产者：
kafka-console-producer --broker-list node1:9092 --topic test
 
修改topic的分区数
kafka-topics --zookeeper node1:2181 --alter --partitions 3 --topic test

消费者：
旧版：kafka-console-consumer --zookeeper node1:2181 --topic test --from-beginning

新版：新版中仍然使用旧版的消费者命令会报错：zookeeper is not a recognized option
        kafka-console-consumer --bootstrap-server node1:9092 --topic test --from-beginning
        指定消费者组：kafka-console-consumer --bootstrap-server node1:9092 --group testGroup --topic test --from-beginning

删除topic
kafka-topics --delete --zookeeper node1:2181 --topic test

查看消费位置
指定消费者组：
    kafka-consumer-groups --bootstrap-server node1:9092 --describe --group testGroup 
    kafka-consumer-groups --bootstrap-server node1:9092 --describe --group testGroup --all-topics

=============================================

创建topic
kafka-topics --create --zookeeper 10.0.0.181:2181 --replication-factor 3 --partitions 3 --topic test

展示topic
kafka-topics --list --zookeeper 10.0.0.181:2181

生产者：
kafka-console-producer --broker-list 10.0.0.181:9092 --topic test

消费者：
旧版：kafka-console-consumer --zookeeper 10.0.0.181:2181 --topic test --from-beginning
新版：kafka-console-consumer --bootstrap-server 10.0.0.181:9092 --topic test --from-beginning

删除topic
kafka-topics --delete --zookeeper 10.0.0.181:2181 --topic test

==============================================

创建topic：同时指定key，那么生产数据时必须同时输入key和value

创建一个向主题 kafka-action 发送消息的生产者，同时指定每条消息包含有key
kafka-console-producer --broker-list 10.0.0.181:9092 --topic test --property parse.key=true


由于以上命令没有指定消息key与消息载荷值value（payload）之间的分隔符，默认为制表符，如果希望修改分隔符，可通过key.separator指定。
kafka-console-producer --broker-list 10.0.0.181:9092 --topic test --property parse.key=true --property key.separator=' '

数据分发策略:
    如果生产者指定了partition，同时发送消息中有key，可以指定数据发往哪个partition，当ProducerRecord 的构造参数中有partition的时候，就可以发送到对应partition上，
    如new ProducerRecord<String, String>("topic主题名",partition,"key", "value")
    如果生产者没有指定partition，但是发送消息中有key，那么将根据key的hash值的方式发送数据到那个partition分区/分片中

==============================================
            

kafka 0.9 及以上 有了一个大版本变化，
主要有以下几个方面：
  1.kafka-client 不再区分高低api
  2.kafka 消费者偏移量信息 不再单纯的存储在 zookeeper 中， kafka 会自己维护自己的 消费情况。
    对于某些特殊的情况：如 kafka-console-consumer，目前在 0.10.0.1 还是会存储在 zookeeper 中。

我们先启动消费者，再启动生产者， 再通过 bin/kafka-consumer-groups.sh 进行消费偏移量查询，
注意：
    在执行脚本查询的时候，对于新的 kafka 自维护的 偏移量的  消费者来说， 消费者 同时运行 是必须的。
    否则会出现 group.id 指定 的 group 查找不到的错误 ！！！！
由于kafka 消费者记录 group 的消费 偏移量 有两种方式 ：
    1）kafka 自维护 （新）
    2）zookpeer 维护  (旧)  ，已经逐渐被废弃
所以 ，脚本查看  消费偏移量的方式有两种   kafka自维护  / zookeeper维护


kafka 维护 消费偏移量的 情况：
    1. 查看有那些 group ID 正在进行消费： 
        kafka-consumer-groups --bootstrap-server node1:9092 --list
        注意：
            这里面是没有指定 topic 的，所以查看的所有的 topic 的 消费者 的 group.id  的列表。
            注意： 重名的 group.id 只会显示一次 
            
    2.查看指定group.id 的消费者消费情况 
        kafka-consumer-groups --bootstrap-server node1:9092 --group testGroup --describe    

        
zookeeper 维护  消费偏移量的 情况：(kafka 0.9及以后的版本的kafka-consumer-groups命令后面已不带有--zookeeper的参数)
    1. 查看有那些 group ID 正在进行消费：
            kafka-consumer-groups --zookeeper node1:2181 --list
    2.查看指定group.id 的消费者消费情况 
            kafka-consumer-groups --zookeeper node1:2181 --group testGroup --describe

=====================================================================================

zookeeper 的shell 命令窗口
    执行 zookeeper-client 命令
        ls /consumers/testGroup/offsets/test
        get /consumers/testGroup/offsets/test/0
        get /consumers/testGroup/offsets/test/1
        get /consumers/testGroup/offsets/test/2

=====================================================================================

        1.kafka消费者在会保存其消费的进度，也就是offset，存储的位置根据选用的kafka api不同而不同。
        2.使用 zookeeper.connect 代表 使用kafka-0.10.1.X版本之前的消费者API
                    auto.offset.reset 的值为 smallest(最小值) 和 largest(最大值) (offest保存在zk中)
                    根据javaapi来消费，也就是【kafka.javaapi.consumer.ConsumerConnector、kafka.javaapi.consumer.SimpleConsumer】，我们会配置参数【zookeeper.connect】来消费
                    消费者的offset会更新到zookeeper的【consumers/{group}/offsets/{topic}/{partition}】目录下。消费者API会把offset保存在 /consumers/groupID/offsets/topic/分区号 中。
                    ls /consumers/testGroup1/offsets/test：得 分区号 [0, 1, 2]
                    get /consumers/testGroup1/offsets/test/分区号：得 分区中的 offsets偏移量
                                [zk: localhost(CONNECTED) 0] get /kafka/consumers/zoo-consumer-group/offsets/my-topic/0
                                5662
                                cZxid = 0x20006d28a
                                ctime = Wed Apr 12 18:20:51 CST 2017
                                mZxid = 0x30132b0ed
                                mtime = Tue Aug 22 18:53:22 CST 2017
                                pZxid = 0x20006d28a
                                cversion = 0
                                dataVersion = 5758
                                aclVersion = 0
                                ephemeralOwner = 0x0
                                dataLength = 4
                                numChildren = 0


            3.如果是根据kafka默认的api来消费，即【org.apache.kafka.clients.consumer.KafkaConsumer】，我们会配置参数【bootstrap.servers】来消费。
              而其消费者的offset会更新到一个kafka自带的topic【__consumer_offsets】下面，查看当前group的消费进度，则要依靠kafka自带的工具【kafka-consumer-offset-checker】
                    kafka 0.8版本以前：kafka-consumer-offset-checker --zookeeper node1:2181/kafka --group testGroup  --topic test --broker-info
                    kafka 0.8版本以后：kafka-consumer-groups --bootstrap-server node1:9092 --group testGroup --describe    
                            Group           Topic                          Pid Offset          logSize         Lag             Owner
                            test-consumer-group stable-test                    0   601808          601808          0               none
                            test-consumer-group stable-test                    1   602826          602828          2               none
                            test-consumer-group stable-test                    2   602136          602136          0               none

              4.Group : 消费者组
                Topic : topic的名字
                Pid : partition的ID
                Offset : kafka消费者在对应分区上已经消费的消息数【位置】
                logSize : 已经写到该分区的消息数【位置】
                Lag : 还有多少消息未读取（Lag = logSize - Offset）
                Owner : 分区创建在哪个broker

            5.offset更新的方式，不区分是用的哪种api，大致分为两类：
                    1.自动提交，设置enable.auto.commit=true，更新的频率根据参数【auto.commit.interval.ms】来定。这种方式也被称为【at most once】，
                                fetch到消息后就可以更新offset，无论是否消费成功。
                    2.手动提交，设置enable.auto.commit=false，这种方式称为【at least once】。fetch到消息后，等消费完成再调用方法【consumer.commitSync()】，
                                手动更新offset；如果消费失败，则offset也不会更新，此条消息会被重复消费一次。