Kafka 介紹與實踐

1.1 實驗內容

本節課將介紹 Kafka 及實現原理，然後完整搭建，案例演示，學習完本課程，你將對 kafka 有深入的瞭解，很快上手。

1.2 課程來源

參考資料： http://kafka.apache.org/documentation.html

1.3. 實驗知識點

• 生產者/消費者模型
• 單機/叢集的區別
• 設計原理

1.4 實驗環境

• Hadoop 2.6.1
• kafka_2.10-0.8.1.1
• Xfce 終端

1.5 適合人群

本課程屬於中等難度級別，適合具有 hadoop 基礎的使用者，如果對分散式檔案系統 瞭解能夠更好的上手本課程。

二、kafka 入門

2.1 簡介

Kafka 是 linkedin 用於日誌處理的分散式訊息佇列，同時支援離線和線上日誌處理。kafka 對訊息儲存時根據 Topic 進行歸類，傳送訊息者成為 Producer，訊息接受者成為 Consumer，此外 kafka 叢集由多個 kafka 例項組成，每個例項 (server) 稱為 broker。無論是 kafka 叢集，還是 producer 和 consumer 都依賴於 zookeeper 來保證系統可用性，為叢集儲存一些 meta 資訊。

2.2 Kafka應用場景

1. 日誌收集

日誌收集方面，其實開源產品有很多，包括 Scribe、Apache Flume。很多人使用 Kafka 代替日誌聚合（log aggregation）。日誌聚合一般來說是從伺服器上收集日誌檔案，然後放到一個集中的位置（檔案伺服器或 HDFS）進行處理。然而 Kafka忽略掉檔案的細節，將其更清晰地抽象成一個個日誌或事件的訊息流。這就讓 Kafka 處理過程延遲更低，更容易支援多資料來源和分散式資料處理。比起以日誌為中心的系統比如 Scribe 或者 Flume 來說，Kafka 提供同樣高效的效能和因為複製導致的更高的耐用性保證，以及更低的端到端延遲。

2. 行為跟蹤

Kafka 的另一個應用場景是跟蹤使用者瀏覽頁面、搜尋及其他行為，以釋出-訂閱的模式實時記錄到對應的 topic 裡。那麼這些結果被訂閱者拿到後，就可以做進一步的實時處理，或實時監控，或放到 Hadoop 離線資料倉庫裡處理。

3. 永續性日誌（commit log）

Kafka 可以為一種外部的永續性日誌的分散式系統提供服務。這種日誌可以在節點間備份資料，併為故障節點資料回覆提供一種重新同步的機制。Kafka 中日誌壓縮功能為這種用法提供了條件。在這種用法中，Kafka 類似於 Apache BookKeeper 專案。

2.3 Kafka基本概念

Topic：特指 Kafka 處理的訊息源（feeds of messages）的不同分類。

Partition：Topic 物理上的分組，一個 topic 可以分為多個 partition，每個 partition 是一個有序的佇列。partition 中的每條訊息都會被分配一個有序的 id（offset）。

Message：訊息，是通訊的基本單位，每個 producer 可以向一個 topic（主題）釋出一些訊息。

Producers：訊息和資料生產者，向 Kafka 的一個 topic 釋出訊息的過程叫做 producers。

Consumers：訊息和資料消費者，訂閱 topics 並處理其釋出的訊息的過程叫做 consumers。

Broker：快取代理，Kafka 叢集中的一臺或多臺伺服器統稱為 broker。

三、設計原理

Kafka 的設計初衷是希望作為一個統一的資訊收集平臺，能夠實時的收集反饋資訊，並需要能夠支撐較大的資料量，且具備良好的容錯能力。

3.1 Kafka 的 Topics/Log

一個Topic 可以認為是一類訊息，每個 topic 將被分成多 partition (區),每個partition在儲存層面是 append log 檔案。任何釋出到此 partition 的訊息都會被直接追加到 log 檔案的尾部，每條訊息在檔案中的位置稱 offset（偏移量），partition 是以檔案的形式儲存在檔案系統中。Logs 檔案根據 broker 中的配置要求,保留一定時間後刪除來釋放磁碟空間。

Partition：

Topic 物理上的分組，一個 topic 可以分為多個 partition，每個 partition 是一個有序的佇列。 partition 中的每條訊息都會被分配一個有序的 id（offset）。

3.2　Kafka的儲存策略

1. kafka以 topic 來進行訊息管理，每個 topic 包含多個 partition，每個 part 對應一個邏輯 log，有多個 segment 組成。

2. 每個 segment 中儲存多條訊息（見下圖），訊息 id 由其邏輯位置決定，即從訊息id可直接定位到訊息的儲存位置，避免id到位置的額外對映。

3. broker 收到釋出訊息往對應 partition 的最後一個 segment 上新增該訊息。

4. 每個 part 在記憶體中對應一個index，記錄每個 segment 中的第一條訊息偏移。

5. 釋出者發到某個 topic 的 訊息會被均勻的分佈到多個 part 上（隨機或根據使用者指定的回撥函式進行分佈），broker 收到釋出訊息往對應 part 的最後一個 segment 上新增 該訊息，當某個 segment 上的訊息條數達到配置值或訊息釋出時間超過閾值時，segment 上的訊息會被 flush 到磁碟，只有 flush 到磁碟上的訊息訂閱者才能訂閱到，segment 達到一定的大小後將不會再往該 segment 寫資料，broker 會建立新的 segment。

3.3 Kafka的訊息傳送的流程

由於 kafka broker 會持久化資料，broker 沒有記憶體壓力，因此，consumer 非常適合採取 pull 的方式消費資料 Producer 向 Kafka（push）推資料，consumer 從 kafka 拉（pull）資料。

3.4 Kafka 的 Zookeeper 協調控制

1. 管理 broker 與 consumer 的動態加入與離開。
2. 觸發負載均衡，當 broker 或 consumer 加入或離開時會觸發負載均衡演算法，使得一個 consumer group 內的多個 consumer 的訂閱負載平衡。
3. 維護消費關係及每個 partion 的消費資訊。

Zookeeper上的細節：

1. 每個 broker 啟動後會在 zookeeper 上註冊一個臨時的 broker registry，包含 broker 的 ip 地址和埠號，所儲存的 topics 和 partitions 資訊。

2. 每個 consumer 啟動後會在 zookeeper 上註冊一個臨時的 consumer registry：包含 consumer 所屬的consumer group 以及訂閱的 topics。

3. 每個 consumer group 關 聯一個臨時的 owner registry 和一個持久的 offset registry。對於被訂閱的每個 partition 包含一個 owner registry，內容為訂閱這個 partition 的 consumer id；同時包含一個 offset registry，內容為上一次訂閱的 offset。

四、安裝部署

叢集方式：單節點單 broker，單節點多 broker，多節點多 broker。

4.1　準備工作

我們已經在實驗樓環境裡下載並配置啟動 hadoop-2.6.1 所需的檔案，免除您配置檔案的麻煩，您可以在 /opt 找到，只需格式化並啟動 hadoop 程序即可。

雙擊開啟桌面上的 Xfce 終端，用 sudo 命令切換到 hadoop 使用者，hadoop 使用者密碼為 hadoop，用 cd 命令進入 /opt目錄。

$ su hadoop$ cd /opt/

在 /opt 目錄下格式化 hadoop。

$ hadoop-2.6.1/bin/hdfs namenode -format

在 /opt 目錄下啟動 hadoop 程序。

$ hadoop-2.6.1/sbin/start-all.sh

用 jps 檢視 hadoop 程序是否啟動。

4.2　下載 kafka 及解壓

你可以通過下面命令將 Kafka 下載到實驗樓環境中，作為參照對比進行學習。

$ sudo wget http://labfile.oss.aliyuncs.com/courses/785/kafka_2.10-0.8.1.1.tgz

在 /opt 目錄下，用 tar 命令解壓下載的 Kafka。

4.3　單節點單 broker

Kafka 用到了 Zookeeper，所有首先啟動 Zookeeper，下面簡單的啟用一個單例項的 Zookeeper 服務。可以在命令的結尾加個 & 符號，這樣就可以啟動後離開控制檯。

注意：kafka 自帶了 zookeeper,為了方便，這裡我們直接使用，當然也可以使用自己下載的 zookeeper*.tar.gz

#許可權不足，授權[email protected]:/opt$ sudo chmod 777 -R kafka_2.10-0.8.1.1#啟動自帶的zookeeper[email protected]:/home/shiyanlou$ cd /opt/kafka_2.10-0.8.1.1/[email protected]:/opt/kafka_2.10-0.8.1.1$  bin/zookeeper-server-start.sh  config/zookeeper.properties &#檢視程序

啟動 kafka 伺服器。

[email protected]:/opt/kafka_2.10-0.8.1.1$  bin/kafka-server-start.sh  config/server.properties  >/dev/null 2>&1 &

檢視 kafka 程序。

建立一個叫做“test”的 topic，它只有一個分割槽，一個副本。

[email protected]:/opt/kafka_2.10-0.8.1.1$  bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

可以通過 list 命令檢視建立的 topic:

檢視一個 topic 的分割槽及副本狀態資訊。

啟動 producer:執行 producer 並在控制檯中輸一些訊息，這些訊息將被髮送到服務端。

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

啟動 consumer:接著再開啟一個終端,執行 consumer 可以讀取訊息並輸出到標準輸出。

hado[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

邊生產邊消費。

4.4　單節點多 broker

4.3節只是啟動了單個 broker，現在啟動由3個 broker 組成的叢集，這些 broker 節點也都是在本機上的，首先為每個節點編寫配置檔案。

[email protected]:/opt/kafka_2.10-0.8.1.1$ cp config/server.properties  config/server-1.properties [email protected]:/opt/kafka_2.10-0.8.1.1$ cp config/server.properties  config/server-2.properties [email protected]:/opt/kafka_2.10-0.8.1.1$

注意: broker.id 在叢集中唯一的標註一個節點，因為在同一個機器上，所以必須制定不同的埠和日誌檔案，避免資料被覆蓋。

在拷貝出的新檔案中新增以下引數：config/server-1.properties:    broker.id=1    port=9093    log.dir=/tmp/kafka-logs-1config/server-2.properties:    broker.id=2    port=9094    log.dir=/tmp/kafka-logs-2

[email protected]:/opt/kafka_2.10-0.8.1.1$ sudo vi config/server-1.properties

[email protected]:/opt/kafka_2.10-0.8.1.1$ sudo vi config/server-2.properties

剛才已經啟動一個 Zookeeper 和一個節點，現在啟動另外兩個節點，只需要在開啟的兩個終端做同樣的操作，如下：

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-server-start.sh config/server-1.properties &[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-server-start.sh config/server-2.properties &

建立一個擁有3個副本的 topic:

[email protected]:/opt/kafka_2.10-0.8.1.1$bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic test2

單節點多 broker 搭建完畢，那麼現在怎麼知道每個節點的資訊呢？執行“describe topics”命令就可以。

[email protected]:/opt/kafka_2.10-0.8.1.1$bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test2

下面解釋一下這些輸出，第一行是對所有分割槽的一個描述，然後每個分割槽都會對應一行，因為我們只有一個分割槽所以下面就只加了一行。

leader：負責處理訊息的讀和寫，leader 是從所有節點中隨機選擇的。

replicas：列出了所有的副本節點，不管節點是否在服務中。

isr：是正在服務中的節點。

在我們的例子中，節點2是作為 leader 執行。啟動生產者向 topic 傳送訊息：

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test2

生產者生產訊息。

啟動消費者，接收訊息。

[email protected]:/opt/kafka_2.10-0.8.1.1$  bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic test2

消費者消費訊息。

4.5 多節點 broker

介紹了上面兩種配置方法，再理解叢集配置就簡單了，zookeeper 配置檔案（zookeeper.properties）不變

broker 的配置配置檔案(server.properties)：按照單節點多例項配置方法在一個節點上啟動一個或多個例項，不同的地方是 zookeeper 的連線串需要把所有節點的 zookeeper 都連線起來，然後複製 kafka 到從機，在所有節點啟動 Kafka broker 即可。

注意：由於實驗環境的限制，多節點 broker 在此不演示了，有條件的同學可以課下練習。

可能遇到的問題。

。

解決辦法。

首先檢查server.properties檔案的broker.id及埠號是否唯一，

必要時用 kill 命令殺死 kafka 程序，重啟 kafka 伺服器。

五、實驗總結

本課程首先簡單講解 kafka 理論，接著講解安裝的三種模式，並實踐操作，對深入學習 kafka 有很大幫助。

六、參考文獻

• http://kafka.apache.org/documentation.html。
• http://blog.sina.com.cn/s/blog_5c51172c0102uxb0.html

           

再分享一下我老師大神的人工智慧教程吧。零基礎！通俗易懂！風趣幽默！還帶黃段子！希望你也加入到我們人工智慧的隊伍中來！https://blog.csdn.net/jiangjunshow