【Kafka】Kafka 架構深入

Kafka 工作流程及檔案儲存機制

Kafka 中訊息是以 topic 進行分類的，生產者生產訊息，消費者消費訊息，都是面向 topic 的。

topic 是邏輯上的概念，而 partition 是物理上的概念，每個 partition 對應於一個 log 檔案，該 log 檔案中儲存的就是 producer 生產的資料。Producer 生產的資料會被不斷追加到該 log 檔案末端，且每條資料都有自己的 offset。消費者組中的每個消費者，都會即時記錄自己消費到了哪個 offset，以便出錯恢復時，從上次的位置繼續消費。

由於生產者生產的訊息會不斷追加到 log 檔案末尾，為防止 log 檔案過大導致資料定位效率低下，Kafka 採取了分片和索引機制，將每個 partition 分為多個 segment。每個 segment 對應兩個檔案：“.index” 檔案和 “.log” 檔案。這些檔案位於一個資料夾下，該資料夾的命名規則為：topic名稱+分割槽序號。

例如，test 這個 topic 有三個分割槽，則其對應的資料夾為 test-0、test-1、test-2。

index 和 log 檔案以當前 segment 的第一條訊息的 offset 命名

“.index” 檔案儲存大量的索引資訊，“.log” 檔案儲存大量的資料，索引檔案中的元資料指向對應資料檔案中 message 的物理偏移地址

資料可靠性保證

為保證 producer 傳送的資料，能可靠的傳送到指定的 topic，topic 的每個 partition 收到 producer 傳送的資料後，都需要向 producer 傳送 ack（acknowledgement 確認收到），如果 producer 收到 ack，就會進行下一輪的傳送，否則重新發送資料。

資料一致性問題

LEO：指的是每個副本最大的 offset； HW：指的是消費者能見到的最大的 offset，所有副本中最小的 LEO

1）follower 故障

follower 發生故障後會被臨時踢出 ISR（Leader 維護的一個和 Leader 保持同步的 Follower 集合），待該 follower 恢復後，follower 會讀取本地磁碟記錄的上次的 HW，並將 log 檔案高於 HW 的部分擷取掉，從 HW 開始向 leader 進行同步。等該 follower 的 LEO 大於等於該 Partition 的 HW，即 follower 追上 leader 之後，就可以重新加入 ISR 了。

2）leader 故障

leader 發生故障之後，會從 ISR 中選出一個新的 leader，之後，為保證多個副本之間的資料一致性，其餘的 follower 會先將各自的 log 檔案高於 HW 的部分截掉，然後從新的 leader 同步資料。

注：這隻能保證副本之間的資料一致性，並不能保證資料不丟失或者不重複。

ack 應答機制

對於某些不太重要的資料，對資料的可靠性要求不是很高，能夠容忍資料的少量丟失，所以沒必要等 ISR 中的 follower 全部接收成功。所以 Kafka 為使用者提供了三種可靠性級別，使用者根據對可靠性和延遲的要求進行權衡選擇。

當 producer 向 leader 傳送資料時，可以透過 request.required.acks 引數來設定資料可靠性的級別：●0：這意味著producer無需等待來自broker的確認而繼續傳送下一批訊息。這種情況下資料傳輸效率最高，但是資料可靠性確是最低的。當broker故障時有可能丟失資料。

●1（預設配置）：這意味著producer在ISR中的leader已成功收到的資料並得到確認後傳送下一條message。如果在follower同步成功之前leader故障，那麼將會丟失資料。

●-1（或者是all）：producer需要等待ISR中的所有follower都確認接收到資料後才算一次傳送完成，可靠性最高。但是如果在 follower 同步完成後，broker 傳送ack 之前，leader 發生故障，那麼會造成資料重複。

三種機制效能依次遞減，資料可靠性依次遞增。

注：在 0.11 版本以前的Kafka，對此是無能為力的，只能保證資料不丟失，再在下游消費者對資料做全域性去重。在 0.11 及以後版本的 Kafka，引入了一項重大特性：冪等性。所謂的冪等性就是指 Producer 不論向 Server 傳送多少次重複資料， Server 端都只會持久化一條。

Filebeat+Kafka+ELK

確保node1 上有安裝apache服務來產生日誌

環境準備

node1：192.168.67.11        elasticsearch  kibananode2：192.168.67.12        elasticsearchapache：192.168.67.10               logstash  apache/nginx/mysqlFilebeat節點：filebeat/192.168.67.13           Filebeatzk-kfk01：192.168.67.21                zookeeper、kafkazk-kfk02：192.168.67.22                zookeeper、kafkazk-kfk03：192.168.67.23                zookeeper、kafkasystemctl stop firewalldsystemctl enable firewalldsetenforce 0

1、部署 Zookeeper+Kafka 叢集

重啟服務

systemctl restart elasticsearch.servicenetstat -antp | grep 9200cd /usr/local/src/elasticsearch-head/npm run start &

2、部署 Filebeat

cd /etc/filebeat#cd /usr/local/filebeatvim filebeat.ymlfilebeat.prospectors:- type: log  enabled: true  paths:    - /var/log/httpd/access_log  tags: ["access"]- type: log  enabled: true  paths:    - /var/log/httpd/error_log  tags: ["error"]......#新增輸出到 Kafka 的配置output.kafka:  enabled: true  #指定 Kafka 叢集配置  hosts: ["192.168.67.21:9092","192.168.67.22:9092","192.168.67.23:9092"]  #指定 Kafka 的 topic  topic: "httpd"

註釋掉logstash 出口，留下kafka出口；出口只能有一個

啟動 filebeat

systemctl restart filebeat.servicesystemctl status filebeat.service# ./filebeat -e -c filebeat.yml

報錯：服務起不來；檢視日誌；

原因：是filebeat.yml中將日誌同時輸出到了kafka和logstash

解決：註釋掉logstash即可

3、部署 ELK，在 Logstash 元件所在節點上新建一個 Logstash 配置檔案

cd /etc/logstash/conf.d/vim kafka.confinput {    kafka {        #kafka叢集地址        bootstrap_servers => "192.168.67.21:9092,192.168.67.22:9092,192.168.67.23:9092"        #拉取的kafka的指定topic        topics  => "httpd"        #指定 type 欄位        type => "httpd_kafka"        #解析json格式的日誌資料        codec => "json"        #拉取最近資料，earliest為從頭開始拉取        auto_offset_reset => "latest"        #傳遞給elasticsearch的資料額外增加kafka的屬性資料        decorate_events => true    }}output {  if "access" in [tags] {    elasticsearch {      hosts => ["192.168.67.11:9200"]      index => "httpd_access-%{+YYYY.MM.dd}"    }  }  if "error" in [tags] {    elasticsearch {      hosts => ["192.168.67.11:9200"]      index => "httpd_error-%{+YYYY.MM.dd}"    }  }  stdout { codec => rubydebug }}