在團800運維工做總結之kafka集羣平常工做經驗總結

1. 一些重要的原理

   基本原理什麼叫broker partition cg我就不在這裏說了，說一些本身總結的原理

   
   

   
   

   1.kafka有副本的概念，每一個副本都分在不一樣的partition中，這中間分爲leader和fllower

   2.kafka消費端的程序必定要和partition數量一致，不能夠多，會出現有些consumer獲取

   不到數據的現象

   3.producer原理

   producer經過zookeeper獲取所鏈接的topic都在那些partiton中，每一個parition的leader是那

   個，針對leader進行寫操做，prodcer經過zookeeper的watch機制來記錄以上的信息，pro

   ducer爲了節省網絡的io，還會在本地先把消息buffer起來，並將他們批量發送到broker中

   4.consumer原理

   consumer向broker發送fetch請求，並告知獲取的消息offset，在kafka中採用pull方式，消費端

   主動pull消息，優勢:消費者能夠控制消費的數量

2.kafka生產環境經常使用命令總結

  

1.模擬生產端，推送數據

./bin/kafka-console-producer.sh --broker-list 172.16.10.130:9092 --topic deal_exposure_origin

2.模擬消費端，消費數據

./bin/kafka-console-consumer.sh --zookeeper 1172.16.10.140:2181  --topic deal_exposure_origin

3.建立topic，topic partiton數量 副本數 數據過時時間

./kafka-topics.sh --zookeeper spark:2181 --create --topic deal_task_log  --partitions 15 --replication-factor 1 retention.ms 1296000000

3.kafka如何動態的添加副本

1.副本，kafka必定要設置副本，若是以後再加會因爲涉及到數據的同步，會把集羣的io提高上去

3.如何擴大副本

2.把全部topic的信息記錄到json文件中，信息有topic名稱，用了哪些partition，副本在那個partition，

並修改json數據，添加副本數

#!/usr/bin/python

from kazoo.client import KazooClient

import random

import json

zk = KazooClient(hosts='172.16.11.73:2181')

zk.start()

for i in zk.get_children('/brokers/topics'):

        b= zk.get('/brokers/topics/'+i)[0]

        a = eval(b)['partitions']

        list = []

        dict = {}

        for key,value in a.items():

                if len(value) == 1:

                        c = {}

                        c['topic'] = i.encode('utf-8')

                        c['partition'] = int(key)

                        list1 = []

                        for ii in range(0,3):

                                while True:

                                        if list1:

                                                pass

                                        else:

                                                for iii in value:

                                                        list1.append(iii)

                                        if len(list1) == 3:

                                                break

                                        num = random.randint(0,4)

                                        #print 'num='+str(num),'value='+str(value)

                                        if num not in list1:

                                                list1.append(num)

                        #print list1

                        c['replicas'] = list1

                        list.append(c)

        version = eval(b)['version']

        dict['version'] = version

        dict['partitions'] = list

        #jsondata = json.dumps(dict)

        json.dump(dict,open('/opt/json/'+i+'.json','w'))

3.加載json文件

/usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-reassign-partitions.sh --zookeeper 192.168.5.159:2181 --reassignment-json-file /opt/test.json --execute

4.查看是否已經添加了副本

usr/local/kafka_2.9.2-0.8.1.1/bin/kafka-topics.sh --describe --zookeeper 192.168.5.159:2181 --topic testtest

Topic:testtest  PartitionCount:15       ReplicationFactor:2     Configs:

        Topic: testtest Partition: 0    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 1    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 2    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 3    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 4    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 5    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 6    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 7    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 8    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 9    Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 10   Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 11   Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 12   Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 13   Leader: 0       Replicas: 0,1   Isr: 0,1

        Topic: testtest Partition: 14   Leader: 0       Replicas: 0,1   Isr: 0,1

4.kafka集羣之間作數據同步

找一個broker節點進行同步

1.建立配置文件mirror_consumer.config

配置文件裏寫本地的kafka集羣zookeeper

定義一個group用來去消費全部的topic，進行同步

zookeeper.connect=172.16.11.43:2181,172.16.11.46:2181,172.16.11.60:2181,172.16.11.67:2181,172.16.11.73:2181

group.id=backup-mirror-consumer-group

2.建立配置文件mirror_producer.config

zookeeper，kafka ip寫對端集羣的ip

zookeeper.connect=172.17.1.159:2181,172.17.1.160:2181

metadata.broker.list=172.17.1.159:9092,172.17.1.160:9092

3.同步命令

$KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.MirrorMaker --consumer.config sourceClusterConsumer.config --num.streams 2 --producer.config targetClusterProducer.config --whitelist=".*"

參數詳解

1. 白名單(whitelist) 黑名單(blacklist)

mirror-maker接受精確指定同步topic的白名單和黑名單。使用java標準的正則表達式，爲了方便，逗號(‘,’)被編譯爲java正則中的(‘|’)。

2. Producer timeout

爲了支持高吞吐量，你最好使用異步的內置producer，並將內置producer設置爲阻塞模式（queue.enqueueTimeout.ms=-1）。這樣能夠保證數據(messages)不會丟失。不然，異步producer默認的 enqueueTimeout是0，若是producer內部的隊列滿了，數據(messages)會被丟棄，並拋出QueueFullExceptions異常。而對於阻塞模式的producer，若是內部隊列滿了就會一直等待，從而有效的節制內置consumer的消費速度。你能夠打開producer的的trace logging，隨時查看內部隊列剩餘的量。若是producer的內部隊列長時間處於滿的狀態，這說明對於mirror-maker來講，將消息從新推到目標Kafka集羣或者將消息寫入磁盤是瓶頸。

對於kafka的producer同步異步的詳細配置請參考$KAFKA_HOME/config/producer.properties文件。關注其中的producer.type和queue.enqueueTimeout.ms這兩個字段。

3. Producer 重試次數（retries）

若是你在producer的配置中使用broker.list，你能夠設置當發佈數據失敗時候的重試次數。retry參數只在使用broker.list的時候使用，由於在重試的時候會從新選擇broker。

4. Producer 數量

經過設置—num.producers參數，可使用一個producer池來提升mirror maker的吞吐量。在接受數據(messages)的broker上的producer是隻使用單個線程來處理的。就算你有多個消費流，吞吐量也會在producer處理請求的時候被限制。

5. 消費流（consumption streams）數量

使用—num.streams能夠指定consumer的線程數。請注意，若是你啓動多個mirror maker進程，你可能須要看看其在源Kafka集羣partitions的分佈狀況。若是在每一個mirror maker進程上的消費流（consumption streams）數量太多，某些消費進程若是不擁有任何分區的消費權限會被置於空閒狀態，主要緣由在於consumer的負載均衡算法。

6. 淺迭代（Shallow iteration）與producer壓縮

咱們建議在mirror maker的consumer中開啓淺迭代（shallow iteration)。意思就是mirror maker的consumer不對已經壓縮的消息集（message-sets）進行解壓，只是直接將獲取到的消息集數據同步到producer中。

若是你開啓淺迭代（shallow iteration)，那麼你必須關閉mirror maker中producer的壓縮功能，不然消息集（message-sets）會被重複壓縮。

7. Consumer 和 源Kafka集羣（source cluster）的 socket buffer sizes

鏡像常常用在跨集羣場景中，你可能但願經過一些配置選項來優化內部集羣的通訊延遲和特定硬件性能瓶頸。通常來講，你應該對mirror-maker中consumer的socket.buffersize 和源集羣broker的socket.send.buffer設定一個高的值。此外，mirror-maker中消費者（consumer）的fetch.size應該設定比socket.buffersize更高的值。注意，套接字緩衝區大小（socket buffer size）是操做系統網絡層的參數。若是你啓用trace級別的日誌，你能夠檢查實際接收的緩衝區大小（buffer size），以肯定是否調整操做系統的網絡層。

4.如何檢驗MirrorMaker運行情況

Consumer offset checker工具能夠用來檢查鏡像對源集羣的消費進度。例如：

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group KafkaMirror --zkconnect localhost:2181 --topic test-topic

KafkaMirror,topic1,0-0 (Group,Topic,BrokerId-PartitionId)

            Owner = KafkaMirror_jkoshy-ld-1320972386342-beb4bfc9-0

  Consumer offset = 561154288

                  = 561,154,288 (0.52G)

         Log size = 2231392259

                  = 2,231,392,259 (2.08G)

     Consumer lag = 1670237971

                  = 1,670,237,971 (1.56G)

BROKER INFO

0 -> 127.0.0.1:9092

注意，–zkconnect參數須要指定到源集羣的Zookeeper。另外，若是指定topic沒有指定，則打印當前消費者group下全部topic的信息。

5.kafka所使用的磁盤io太高解決方法

問題:kafka所用磁盤io太高

咱們生產平臺有5臺kafka機器,每臺機器上分了2塊磁盤作parition

最近發現kafka所使用的磁盤io很是高，影響到了生產端推送數據的性能

一開始覺得是因爲一個推送日誌的topic所致使的的，由於每秒推送數據大概在2w左右，

後來把此topic遷移到了其餘的kafka集羣中仍是未見效果

最終iotop發現實際上是因爲zookeeper持久化的時候致使的

zookeeper持久化的時候也寫到kafka所用到的磁盤中

經過此問題說明幾點問題

1.kafka用zookeeper，和你們所熟悉的其餘應用例如solrcloud codis otter不太同樣

通常用zookeeper都是管理集羣節點用，而kafka用zookeeper是核心，生產端和消費端都會去

連接zookeeper獲取響應的信息

生產端經過連接zookeeper獲取topic都用了那些parition，每一個parition的副本的leader是那個

消費端連接zookeeper獲取offset，消費端消費都會操做對zookeeper的數據進行修改，對io的操做

很頻繁

解決方法：

禁止zookeeper作持久化操做

配置文件中添加一行

forceSync=no

問題解決