一 Kafka概述
1.1 Kafka是什么
在流式計算中，Kafka一般用來緩存數據，Storm通過消費Kafka的數據進行計算。
1）Apache Kafka是一個開源消息系統，由Scala寫成。是由Apache軟件基金會開發的一個開源消息系統項目。
2）Kafka最初是由LinkedIn開發，並於2011年初開源。2012年10月從Apache Incubator畢業。該項目的目標是為處理實時數據提供一個統一、高通量、低等待的平台。
3）Kafka是一個分布式消息隊列。Kafka對消息保存時根據Topic進行歸類，發送消息者稱為Producer，消息接受者稱為Consumer，此外kafka集群有多個kafka實例組成，每個實例(server)成為broker。
4）無論是kafka集群，還是producer和consumer都依賴於zookeeper集群保存一些meta信息，來保證系統可用性。
1.2 Kafka內部實現原理
（1）點對點模式（一對一，消費者主動拉取數據，消息收到后消息清除）
點對點模型通常是一個基於拉取或者輪詢的消息傳送模型，這種模型從隊列中請求信息，而不是將消息推送到客戶端。這個模型的特點是發送到隊列的消息被一個且只有一個接收者接收處理，即使有多個消息監聽者也是如此。
（2）發布/訂閱模式（一對多，數據生產后，推送給所有訂閱者）
發布訂閱模型則是一個基於推送的消息傳送模型。發布訂閱模型可以有多種不同的訂閱者，臨時訂閱者只在主動監聽主題時才接收消息，而持久訂閱者則監聽主題的所有消息，即使當前訂閱者不可用，處於離線狀態。
1.2 為什么需要消息隊列
1）解耦：
允許你獨立的擴展或修改兩邊的處理過程，只要確保它們遵守同樣的接口約束。
2）冗余：
消息隊列把數據進行持久化直到它們已經被完全處理，通過這一方式規避了數據丟失風險。許多消息隊列所采用的"插入-獲取-刪除"范式中，在把一個消息從隊列中刪除之前，需要你的處理系統明確的指出該消息已經被處理完畢，從而確保你的數據被安全的保存直到你使用完畢。
3）擴展性：
因為消息隊列解耦了你的處理過程，所以增大消息入隊和處理的頻率是很容易的，只要另外增加處理過程即可。
4）靈活性 & 峰值處理能力：
在訪問量劇增的情況下，應用仍然需要繼續發揮作用，但是這樣的突發流量並不常見。如果為以能處理這類峰值訪問為標准來投入資源隨時待命無疑是巨大的浪費。使用消息隊列能夠使關鍵組件頂住突發的訪問壓力，而不會因為突發的超負荷的請求而完全崩潰。
5）可恢復性：
系統的一部分組件失效時，不會影響到整個系統。消息隊列降低了進程間的耦合度，所以即使一個處理消息的進程掛掉，加入隊列中的消息仍然可以在系統恢復后被處理。
6）順序保證：
在大多使用場景下，數據處理的順序都很重要。大部分消息隊列本來就是排序的，並且能保證數據會按照特定的順序來處理。（Kafka保證一個Partition內的消息的有序性）
7）緩沖：
有助於控制和優化數據流經過系統的速度，解決生產消息和消費消息的處理速度不一致的情況。
8）異步通信：
很多時候，用戶不想也不需要立即處理消息。消息隊列提供了異步處理機制，允許用戶把一個消息放入隊列，但並不立即處理它。想向隊列中放入多少消息就放多少，然后在需要的時候再去處理它們。
1.3 Kafka架構
1）Producer ：消息生產者，就是向kafka broker發消息的客戶端。
2）Consumer ：消息消費者，向kafka broker取消息的客戶端
3）Topic ：可以理解為一個隊列。
4） Consumer Group （CG）：這是kafka用來實現一個topic消息的廣播（發給所有的consumer）和單播（發給任意一個consumer）的手段。一個topic可以有多個CG。topic的消息會復制（不是真的復制，是概念上的）到所有的CG，但每個partion只會把消息發給該CG中的一個consumer。如果需要實現廣播，只要每個consumer有一個獨立的CG就可以了。要實現單播只要所有的consumer在同一個CG。用CG還可以將consumer進行自由的分組而不需要多次發送消息到不同的topic。
5）Broker ：一台kafka服務器就是一個broker。一個集群由多個broker組成。一個broker可以容納多個topic。
6）Partition：為了實現擴展性，一個非常大的topic可以分布到多個broker（即服務器）上，一個topic可以分為多個partition，每個partition是一個有序的隊列。partition中的每條消息都會被分配一個有序的id（offset）。kafka只保證按一個partition中的順序將消息發給consumer，不保證一個topic的整體（多個partition間）的順序。
7）Offset：kafka的存儲文件都是按照offset.kafka來命名，用offset做名字的好處是方便查找。例如你想找位於2049的位置，只要找到2048.kafka的文件即可。當然the first offset就是00000000000.kafka
二 Kafka集群部署
2.1 環境准備
2.1.1 集群規划
hadoop102 hadoop103 hadoop104
zk zk zk
kafka kafka kafka
2.1.2 jar包下載
http://kafka.apache.org/downloads.html
2.1.3 虛擬機准備
1）准備3台虛擬機
2）配置ip地址
3）配置主機名稱
4）3台主機分別關閉防火牆
[root@hadoop102 atguigu]# chkconfig iptables off
[root@hadoop103 atguigu]# chkconfig iptables off
[root@hadoop104 atguigu]# chkconfig iptables off
2.1.4 安裝jdk
2.1.5 安裝Zookeeper
0）集群規划
在hadoop102、hadoop103和hadoop104三個節點上部署Zookeeper。
1）解壓安裝
（1）解壓zookeeper安裝包到/opt/module/目錄下
[atguigu@hadoop102 software]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
（2）在/opt/module/zookeeper-3.4.10/這個目錄下創建zkData
mkdir -p zkData
（3）重命名/opt/module/zookeeper-3.4.10/conf這個目錄下的zoo_sample.cfg為zoo.cfg
mv zoo_sample.cfg zoo.cfg
2）配置zoo.cfg文件
（1）具體配置
dataDir=/opt/module/zookeeper-3.4.10/zkData
增加如下配置
#######################cluster##########################
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888
（2）配置參數解讀
Server.A=B:C:D。
A是一個數字，表示這個是第幾號服務器；
B是這個服務器的ip地址；
C是這個服務器與集群中的Leader服務器交換信息的端口；
D是萬一集群中的Leader服務器掛了，需要一個端口來重新進行選舉，選出一個新的Leader，而這個端口就是用來執行選舉時服務器相互通信的端口。
集群模式下配置一個文件myid，這個文件在dataDir目錄下，這個文件里面有一個數據就是A的值，Zookeeper啟動時讀取此文件，拿到里面的數據與zoo.cfg里面的配置信息比較從而判斷到底是哪個server。
3）集群操作
（1）在/opt/module/zookeeper-3.4.10/zkData目錄下創建一個myid的文件
touch myid
添加myid文件，注意一定要在linux里面創建，在notepad++里面很可能亂碼
（2）編輯myid文件
vi myid
在文件中添加與server對應的編號：如2
（3）拷貝配置好的zookeeper到其他機器上
scp -r zookeeper-3.4.10/ root@hadoop103.atguigu.com:/opt/app/
scp -r zookeeper-3.4.10/ root@hadoop104.atguigu.com:/opt/app/
並分別修改myid文件中內容為3、4
（4）分別啟動zookeeper
[root@hadoop102 zookeeper-3.4.10]# bin/zkServer.sh start
[root@hadoop103 zookeeper-3.4.10]# bin/zkServer.sh start
[root@hadoop104 zookeeper-3.4.10]# bin/zkServer.sh start
（5）查看狀態
[root@hadoop102 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
[root@hadoop103 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: leader
[root@hadoop104 zookeeper-3.4.5]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
2.2 Kafka集群部署
1）解壓安裝包
[atguigu@hadoop102 software]$ tar -zxvf kafka_2.11-0.11.0.0.tgz -C /opt/module/
2）修改解壓后的文件名稱
[atguigu@hadoop102 module]$ mv kafka_2.11-0.11.0.0/ kafka
3）在/opt/module/kafka目錄下創建logs文件夾
[atguigu@hadoop102 kafka]$ mkdir logs
4）修改配置文件
[atguigu@hadoop102 kafka]$ cd config/
[atguigu@hadoop102 config]$ vi server.properties
輸入以下內容：
| #broker的全局唯一編號，不能重復 broker.id=0 #刪除topic功能使能 delete.topic.enable=true #處理網絡請求的線程數量 num.network.threads=3 #用來處理磁盤IO的現成數量 num.io.threads=8 #發送套接字的緩沖區大小 socket.send.buffer.bytes=102400 #接收套接字的緩沖區大小 socket.receive.buffer.bytes=102400 #請求套接字的緩沖區大小 socket.request.max.bytes=104857600 #kafka運行日志存放的路徑 log.dirs=/opt/module/kafka/logs #topic在當前broker上的分區個數 num.partitions=1 #用來恢復和清理data下數據的線程數量 num.recovery.threads.per.data.dir=1 #segment文件保留的最長時間，超時將被刪除 log.retention.hours=168 #配置連接Zookeeper集群地址 zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181 |
5）配置環境變量
[root@hadoop102 module]# vi /etc/profile
| #KAFKA_HOME export KAFKA_HOME=/opt/module/kafka export PATH=$PATH:$KAFKA_HOME/bin |
[root@hadoop102 module]# source /etc/profile
6）分發安裝包
[root@hadoop102 etc]# xsync profile
[atguigu@hadoop102 module]$ xsync kafka/
7）分別在hadoop103和hadoop104上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=1、broker.id=2
注：broker.id不得重復
8）啟動集群
依次在hadoop102、hadoop103、hadoop104節點上啟動kafka
[atguigu@hadoop102 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop103 kafka]$ bin/kafka-server-start.sh config/server.properties &
[atguigu@hadoop104 kafka]$ bin/kafka-server-start.sh config/server.properties &
2.3 Kafka命令行操作
1）查看當前服務器中的所有topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --list --zookeeper hadoop102:2181
2）創建topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --create --zookeeper hadoop102:2181 --replication-factor 3 --partitions 1 --topic first
選項說明：
--topic 定義topic名
--replication-factor 定義副本數
--partitions 定義分區數
3） 刪除topic
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --delete --zookeeper hadoop102:2181 --topic first
需要server.properties中設置delete.topic.enable=true否則只是標記刪除或者直接重啟。
4）發送消息
[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first
>hello world
>atguigu atguigu
5）消費消息
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --from-beginning --topic first
6）查看某個Topic的詳情
[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --topic first --describe --zookeeper hadoop102:2181
2.4 Kafka配置信息
2.4.1 Broker配置信息
| 屬性 |
默認值 |
描述 |
| broker.id |
|
必填參數，broker的唯一標識 |
| log.dirs |
/tmp/kafka-logs |
Kafka數據存放的目錄。可以指定多個目錄，中間用逗號分隔，當新partition被創建的時會被存放到當前存放partition最少的目錄。 |
| port |
9092 |
BrokerServer接受客戶端連接的端口號 |
| zookeeper.connect |
null |
Zookeeper的連接串，格式為：hostname1:port1,hostname2:port2,hostname3:port3。可以填一個或多個，為了提高可靠性，建議都填上。注意，此配置允許我們指定一個zookeeper路徑來存放此kafka集群的所有數據，為了與其他應用集群區分開，建議在此配置中指定本集群存放目錄，格式為：hostname1:port1,hostname2:port2,hostname3:port3/chroot/path 。需要注意的是，消費者的參數要和此參數一致。 |
| message.max.bytes |
1000000 |
服務器可以接收到的最大的消息大小。注意此參數要和consumer的maximum.message.size大小一致，否則會因為生產者生產的消息太大導致消費者無法消費。 |
| num.io.threads |
8 |
服務器用來執行讀寫請求的IO線程數，此參數的數量至少要等於服務器上磁盤的數量。 |
| queued.max.requests |
500 |
I/O線程可以處理請求的隊列大小，若實際請求數超過此大小，網絡線程將停止接收新的請求。 |
| socket.send.buffer.bytes |
100 * 1024 |
The SO_SNDBUFF buffer the server prefers for socket connections. |
| socket.receive.buffer.bytes |
100 * 1024 |
The SO_RCVBUFF buffer the server prefers for socket connections. |
| socket.request.max.bytes |
100 * 1024 * 1024 |
服務器允許請求的最大值， 用來防止內存溢出，其值應該小於 Java heap size. |
| num.partitions |
1 |
默認partition數量，如果topic在創建時沒有指定partition數量，默認使用此值，建議改為5 |
| log.segment.bytes |
1024 * 1024 * 1024 |
Segment文件的大小，超過此值將會自動新建一個segment，此值可以被topic級別的參數覆蓋。 |
| log.roll.{ms,hours} |
24 * 7 hours |
新建segment文件的時間，此值可以被topic級別的參數覆蓋。 |
| log.retention.{ms,minutes,hours} |
7 days |
Kafka segment log的保存周期，保存周期超過此時間日志就會被刪除。此參數可以被topic級別參數覆蓋。數據量大時，建議減小此值。 |
| log.retention.bytes |
-1 |
每個partition的最大容量，若數據量超過此值，partition數據將會被刪除。注意這個參數控制的是每個partition而不是topic。此參數可以被log級別參數覆蓋。 |
| log.retention.check.interval.ms |
5 minutes |
刪除策略的檢查周期 |
| auto.create.topics.enable |
true |
自動創建topic參數，建議此值設置為false，嚴格控制topic管理，防止生產者錯寫topic。 |
| default.replication.factor |
1 |
默認副本數量，建議改為2。 |
| replica.lag.time.max.ms |
10000 |
在此窗口時間內沒有收到follower的fetch請求，leader會將其從ISR(in-sync replicas)中移除。 |
| replica.lag.max.messages |
4000 |
如果replica節點落后leader節點此值大小的消息數量，leader節點就會將其從ISR中移除。 |
| replica.socket.timeout.ms |
30 * 1000 |
replica向leader發送請求的超時時間。 |
| replica.socket.receive.buffer.bytes |
64 * 1024 |
The socket receive buffer for network requests to the leader for replicating data. |
| replica.fetch.max.bytes |
1024 * 1024 |
The number of byes of messages to attempt to fetch for each partition in the fetch requests the replicas send to the leader. |
| replica.fetch.wait.max.ms |
500 |
The maximum amount of time to wait time for data to arrive on the leader in the fetch requests sent by the replicas to the leader. |
| num.replica.fetchers |
1 |
Number of threads used to replicate messages from leaders. Increasing this value can increase the degree of I/O parallelism in the follower broker. |
| fetch.purgatory.purge.interval.requests |
1000 |
The purge interval (in number of requests) of the fetch request purgatory. |
| zookeeper.session.timeout.ms |
6000 |
ZooKeeper session 超時時間。如果在此時間內server沒有向zookeeper發送心跳，zookeeper就會認為此節點已掛掉。 此值太低導致節點容易被標記死亡；若太高，.會導致太遲發現節點死亡。 |
| zookeeper.connection.timeout.ms |
6000 |
客戶端連接zookeeper的超時時間。 |
| zookeeper.sync.time.ms |
2000 |
H ZK follower落后 ZK leader的時間。 |
| controlled.shutdown.enable |
true |
允許broker shutdown。如果啟用，broker在關閉自己之前會把它上面的所有leaders轉移到其它brokers上，建議啟用，增加集群穩定性。 |
| auto.leader.rebalance.enable |
true |
If this is enabled the controller will automatically try to balance leadership for partitions among the brokers by periodically returning leadership to the “preferred” replica for each partition if it is available. |
| leader.imbalance.per.broker.percentage |
10 |
The percentage of leader imbalance allowed per broker. The controller will rebalance leadership if this ratio goes above the configured value per broker. |
| leader.imbalance.check.interval.seconds |
300 |
The frequency with which to check for leader imbalance. |
| offset.metadata.max.bytes |
4096 |
The maximum amount of metadata to allow clients to save with their offsets. |
| connections.max.idle.ms |
600000 |
Idle connections timeout: the server socket processor threads close the connections that idle more than this. |
| num.recovery.threads.per.data.dir |
1 |
The number of threads per data directory to be used for log recovery at startup and flushing at shutdown. |
| unclean.leader.election.enable |
true |
Indicates whether to enable replicas not in the ISR set to be elected as leader as a last resort, even though doing so may result in data loss. |
| delete.topic.enable |
false |
啟用deletetopic參數，建議設置為true。 |
| offsets.topic.num.partitions |
50 |
The number of partitions for the offset commit topic. Since changing this after deployment is currently unsupported, we recommend using a higher setting for production (e.g., 100-200). |
| offsets.topic.retention.minutes |
1440 |
Offsets that are older than this age will be marked for deletion. The actual purge will occur when the log cleaner compacts the offsets topic. |
| offsets.retention.check.interval.ms |
600000 |
The frequency at which the offset manager checks for stale offsets. |
| offsets.topic.replication.factor |
3 |
The replication factor for the offset commit topic. A higher setting (e.g., three or four) is recommended in order to ensure higher availability. If the offsets topic is created when fewer brokers than the replication factor then the offsets topic will be created with fewer replicas. |
| offsets.topic.segment.bytes |
104857600 |
Segment size for the offsets topic. Since it uses a compacted topic, this should be kept relatively low in order to facilitate faster log compaction and loads. |
| offsets.load.buffer.size |
5242880 |
An offset load occurs when a broker becomes the offset manager for a set of consumer groups (i.e., when it becomes a leader for an offsets topic partition). This setting corresponds to the batch size (in bytes) to use when reading from the offsets segments when loading offsets into the offset manager’s cache. |
| offsets.commit.required.acks |
-1 |
The number of acknowledgements that are required before the offset commit can be accepted. This is similar to the producer’s acknowledgement setting. In general, the default should not be overridden. |
| offsets.commit.timeout.ms |
5000 |
The offset commit will be delayed until this timeout or the required number of replicas have received the offset commit. This is similar to the producer request timeout. |
2.4.2 Producer配置信息
| 屬性 |
默認值 |
描述 |
| metadata.broker.list |
|
啟動時producer查詢brokers的列表，可以是集群中所有brokers的一個子集。注意，這個參數只是用來獲取topic的元信息用，producer會從元信息中挑選合適的broker並與之建立socket連接。格式是：host1:port1,host2:port2。 |
| request.required.acks |
0 |
參見3.2節介紹 |
| request.timeout.ms |
10000 |
Broker等待ack的超時時間，若等待時間超過此值，會返回客戶端錯誤信息。 |
| producer.type |
sync |
同步異步模式。async表示異步，sync表示同步。如果設置成異步模式，可以允許生產者以batch的形式push數據，這樣會極大的提高broker性能，推薦設置為異步。 |
| serializer.class |
kafka.serializer.DefaultEncoder |
序列號類，.默認序列化成 byte[] 。 |
| key.serializer.class |
|
Key的序列化類，默認同上。 |
| partitioner.class |
kafka.producer.DefaultPartitioner |
Partition類，默認對key進行hash。 |
| compression.codec |
none |
指定producer消息的壓縮格式，可選參數為： “none”, “gzip” and “snappy”。關於壓縮參見4.1節 |
| compressed.topics |
null |
啟用壓縮的topic名稱。若上面參數選擇了一個壓縮格式，那么壓縮僅對本參數指定的topic有效，若本參數為空，則對所有topic有效。 |
| message.send.max.retries |
3 |
Producer發送失敗時重試次數。若網絡出現問題，可能會導致不斷重試。 |
| retry.backoff.ms |
100 |
Before each retry, the producer refreshes the metadata of relevant topics to see if a new leader has been elected. Since leader election takes a bit of time, this property specifies the amount of time that the producer waits before refreshing the metadata. |
| topic.metadata.refresh.interval.ms |
600 * 1000 |
The producer generally refreshes the topic metadata from brokers when there is a failure (partition missing, leader not available…). It will also poll regularly (default: every 10min so 600000ms). If you set this to a negative value, metadata will only get refreshed on failure. If you set this to zero, the metadata will get refreshed after each message sent (not recommended). Important note: the refresh happen only AFTER the message is sent, so if the producer never sends a message the metadata is never refreshed |
| queue.buffering.max.ms |
5000 |
啟用異步模式時，producer緩存消息的時間。比如我們設置成1000時，它會緩存1秒的數據再一次發送出去，這樣可以極大的增加broker吞吐量，但也會造成時效性的降低。 |
| queue.buffering.max.messages |
10000 |
采用異步模式時producer buffer 隊列里最大緩存的消息數量，如果超過這個數值，producer就會阻塞或者丟掉消息。 |
| queue.enqueue.timeout.ms |
-1 |
當達到上面參數值時producer阻塞等待的時間。如果值設置為0，buffer隊列滿時producer不會阻塞，消息直接被丟掉。若值設置為-1，producer會被阻塞，不會丟消息。 |
| batch.num.messages |
200 |
采用異步模式時，一個batch緩存的消息數量。達到這個數量值時producer才會發送消息。 |
| send.buffer.bytes |
100 * 1024 |
Socket write buffer size |
| client.id |
“” |
The client id is a user-specified string sent in each request to help trace calls. It should logically identify the application making the request. |
2.4.3 Consumer配置信息
| 屬性 |
默認值 |
描述 |
| group.id |
|
Consumer的組ID，相同goup.id的consumer屬於同一個組。 |
| zookeeper.connect |
|
Consumer的zookeeper連接串，要和broker的配置一致。 |
| consumer.id |
null |
如果不設置會自動生成。 |
| socket.timeout.ms |
30 * 1000 |
網絡請求的socket超時時間。實際超時時間由max.fetch.wait + socket.timeout.ms 確定。 |
| socket.receive.buffer.bytes |
64 * 1024 |
The socket receive buffer for network requests. |
| fetch.message.max.bytes |
1024 * 1024 |
查詢topic-partition時允許的最大消息大小。consumer會為每個partition緩存此大小的消息到內存，因此，這個參數可以控制consumer的內存使用量。這個值應該至少比server允許的最大消息大小大，以免producer發送的消息大於consumer允許的消息。 |
| num.consumer.fetchers |
1 |
The number fetcher threads used to fetch data. |
| auto.commit.enable |
true |
如果此值設置為true，consumer會周期性的把當前消費的offset值保存到zookeeper。當consumer失敗重啟之后將會使用此值作為新開始消費的值。 |
| auto.commit.interval.ms |
60 * 1000 |
Consumer提交offset值到zookeeper的周期。 |
| queued.max.message.chunks |
2 |
用來被consumer消費的message chunks 數量， 每個chunk可以緩存fetch.message.max.bytes大小的數據量。 |
| auto.commit.interval.ms |
60 * 1000 |
Consumer提交offset值到zookeeper的周期。 |
| queued.max.message.chunks |
2 |
用來被consumer消費的message chunks 數量， 每個chunk可以緩存fetch.message.max.bytes大小的數據量。 |
| fetch.min.bytes |
1 |
The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request. |
| fetch.wait.max.ms |
100 |
The maximum amount of time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy fetch.min.bytes. |
| rebalance.backoff.ms |
2000 |
Backoff time between retries during rebalance. |
| refresh.leader.backoff.ms |
200 |
Backoff time to wait before trying to determine the leader of a partition that has just lost its leader. |
| auto.offset.reset |
largest |
What to do when there is no initial offset in ZooKeeper or if an offset is out of range ;smallest : automatically reset the offset to the smallest offset; largest : automatically reset the offset to the largest offset;anything else: throw exception to the consumer |
| consumer.timeout.ms |
-1 |
若在指定時間內沒有消息消費，consumer將會拋出異常。 |
| exclude.internal.topics |
true |
Whether messages from internal topics (such as offsets) should be exposed to the consumer. |
| zookeeper.session.timeout.ms |
6000 |
ZooKeeper session timeout. If the consumer fails to heartbeat to ZooKeeper for this period of time it is considered dead and a rebalance will occur. |
| zookeeper.connection.timeout.ms |
6000 |
The max time that the client waits while establishing a connection to zookeeper. |
| zookeeper.sync.time.ms |
2000 |
How far a ZK follower can be behind a ZK leader |
三 Kafka工作流程分析
3.1 Kafka生產過程分析
3.1.1 寫入方式
producer采用推（push）模式將消息發布到broker，每條消息都被追加（append）到分區（patition）中，屬於順序寫磁盤（順序寫磁盤效率比隨機寫內存要高，保障kafka吞吐率）。
3.1.2 分區（Partition）
消息發送時都被發送到一個topic，其本質就是一個目錄，而topic是由一些Partition Logs(分區日志)組成，其組織結構如下圖所示：
我們可以看到，每個Partition中的消息都是有序的，生產的消息被不斷追加到Partition log上，其中的每一個消息都被賦予了一個唯一的offset值。
1）分區的原因
（1）方便在集群中擴展，每個Partition可以通過調整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應任意大小的數據了；
（2）可以提高並發，因為可以以Partition為單位讀寫了。
2）分區的原則
（1）指定了patition，則直接使用；
（2）未指定patition但指定key，通過對key的value進行hash出一個patition
（3）patition和key都未指定，使用輪詢選出一個patition。
| DefaultPartitioner類 public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); if (keyBytes == null) { int nextValue = nextValue(topic); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (availablePartitions.size() > 0) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); return availablePartitions.get(part).partition(); } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } else { // hash the keyBytes to choose a partition return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } } |
3.1.3 副本（Replication）
同一個partition可能會有多個replication（對應 server.properties 配置中的 default.replication.factor=N）。沒有replication的情況下，一旦broker 宕機，其上所有 patition 的數據都不可被消費，同時producer也不能再將數據存於其上的patition。引入replication之后，同一個partition可能會有多個replication，而這時需要在這些replication之間選出一個leader，producer和consumer只與這個leader交互，其它replication作為follower從leader 中復制數據。
3.1.4 寫入流程
producer寫入消息流程如下：
1）producer先從zookeeper的 "/brokers/.../state"節點找到該partition的leader
2）producer將消息發送給該leader
3）leader將消息寫入本地log
4）followers從leader pull消息，寫入本地log后向leader發送ACK
5）leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）並向producer發送ACK
3.2 broker 保存消息
3.2.1 存儲方式
物理上把topic分成一個或多個patition（對應 server.properties 中的num.partitions=3配置），每個patition物理上對應一個文件夾（該文件夾存儲該patition的所有消息和索引文件），如下：
[atguigu@hadoop102 logs]$ ll
drwxrwxr-x. 2 atguigu atguigu 4096 8月 6 14:37 first-0
drwxrwxr-x. 2 atguigu atguigu 4096 8月 6 14:35 first-1
drwxrwxr-x. 2 atguigu atguigu 4096 8月 6 14:37 first-2
[atguigu@hadoop102 logs]$ cd first-0
[atguigu@hadoop102 first-0]$ ll
-rw-rw-r--. 1 atguigu atguigu 10485760 8月 6 14:33 00000000000000000000.index
-rw-rw-r--. 1 atguigu atguigu 219 8月 6 15:07 00000000000000000000.log
-rw-rw-r--. 1 atguigu atguigu 10485756 8月 6 14:33 00000000000000000000.timeindex
-rw-rw-r--. 1 atguigu atguigu 8 8月 6 14:37 leader-epoch-checkpoint
3.2.2 存儲策略
無論消息是否被消費，kafka都會保留所有消息。有兩種策略可以刪除舊數據：
1）基於時間：log.retention.hours=168
2）基於大小：log.retention.bytes=1073741824
需要注意的是，因為Kafka讀取特定消息的時間復雜度為O(1)，即與文件大小無關，所以這里刪除過期文件與提高 Kafka 性能無關。
3.2.3 Zookeeper存儲結構
注意：producer不在zk中注冊，消費者在zk中注冊。
3.3 Kafka消費過程分析
kafka提供了兩套consumer API：高級Consumer API和低級API。
3.3.1 高級API
1）高級API優點
高級API 寫起來簡單
不需要去自行去管理offset，系統通過zookeeper自行管理
不需要管理分區，副本等情況，系統自動管理
消費者斷線會自動根據上一次記錄在zookeeper中的offset去接着獲取數據（默認設置1分鍾更新一下zookeeper中存的的offset）
可以使用group來區分對同一個topic 的不同程序訪問分離開來（不同的group記錄不同的offset，這樣不同程序讀取同一個topic才不會因為offset互相影響）
2）高級API缺點
不能自行控制offset（對於某些特殊需求來說）
不能細化控制如分區、副本、zk等
3.3.2 低級API
1）低級 API 優點
能夠開發者自己控制offset，想從哪里讀取就從哪里讀取。
自行控制連接分區，對分區自定義進行負載均衡
對zookeeper的依賴性降低（如：offset不一定非要靠zk存儲，自行存儲offset即可，比如存在文件或者內存中）
2）低級API缺點
太過復雜，需要自行控制offset，連接哪個分區，找到分區leader 等。
3.3.3 消費者組
消費者是以consumer group消費者組的方式工作，由一個或者多個消費者組成一個組，共同消費一個topic。每個分區在同一時間只能由group中的一個消費者讀取，但是多個group可以同時消費這個partition。在圖中，有一個由三個消費者組成的group，有一個消費者讀取主題中的兩個分區，另外兩個分別讀取一個分區。某個消費者讀取某個分區，也可以叫做某個消費者是某個分區的擁有者。
在這種情況下，消費者可以通過水平擴展的方式同時讀取大量的消息。另外，如果一個消費者失敗了，那么其他的group成員會自動負載均衡讀取之前失敗的消費者讀取的分區。
3.3.4 消費方式
consumer采用pull（拉）模式從broker中讀取數據。
push（推）模式很難適應消費速率不同的消費者，因為消息發送速率是由broker決定的。它的目標是盡可能以最快速度傳遞消息，但是這樣很容易造成consumer來不及處理消息，典型的表現就是拒絕服務以及網絡擁塞。而pull模式則可以根據consumer的消費能力以適當的速率消費消息。
對於Kafka而言，pull模式更合適，它可簡化broker的設計，consumer可自主控制消費消息的速率，同時consumer可以自己控制消費方式——即可批量消費也可逐條消費，同時還能選擇不同的提交方式從而實現不同的傳輸語義。
3.3.5 消費者組案例
1）需求：測試同一個消費者組中的消費者，同一時刻只能有一個消費者消費。
2）案例實操
（1）在hadoop102、hadoop103上修改/opt/module/kafka/config/consumer.properties配置文件中的group.id屬性為任意組名。
[atguigu@hadoop103 config]$ vi consumer.properties
group.id=atguigu
（2）在hadoop102、hadoop103上分別啟動消費者
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first --consumer.config config/consumer.properties
（3）在hadoop104上啟動生產者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first
>hello world
（4）查看hadoop102和hadoop103的接收者。
同一時刻只有一個消費者接收到消息。
四 Kafka API實戰
4.1 環境准備
1）在eclipse中創建一個java工程
2）在工程的根目錄創建一個lib文件夾
3）解壓kafka安裝包，將安裝包libs目錄下的jar包拷貝到工程的lib目錄下，並build path。
4）啟動zk和kafka集群，在kafka集群中打開一個消費者
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first
4.2 Kafka生產者Java API
4.2.1 創建生產者（過時的API）
| package com.atguigu.kafka; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig;
public class OldProducer {
@SuppressWarnings("deprecation") public static void main(String[] args) {
Properties properties = new Properties(); properties.put("metadata.broker.list", "hadoop102:9092"); properties.put("request.required.acks", "1"); properties.put("serializer.class", "kafka.serializer.StringEncoder");
Producer<Integer, String> producer = new Producer<Integer,String>(new ProducerConfig(properties));
KeyedMessage<Integer, String> message = new KeyedMessage<Integer, String>("first", "hello world"); producer.send(message ); } } |
4.2.2 創建生產者（新API）
| package com.atguigu.kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord;
public class NewProducer {
public static void main(String[] args) {
Properties props = new Properties(); // Kafka服務端的主機名和端口號 props.put("bootstrap.servers", "hadoop103:9092"); // 等待所有副本節點的應答 props.put("acks", "all"); // 消息發送最大嘗試次數 props.put("retries", 0); // 一批消息處理大小 props.put("batch.size", 16384); // 請求延時 props.put("linger.ms", 1); // 發送緩存區內存大小 props.put("buffer.memory", 33554432); // key序列化 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props); for (int i = 0; i < 50; i++) { producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i)); }
producer.close(); } } |
4.2.3 創建生產者帶回調函數（新API）
| package com.atguigu.kafka; import java.util.Properties; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata;
public class CallBackProducer {
public static void main(String[] args) {
Properties props = new Properties(); // Kafka服務端的主機名和端口號 props.put("bootstrap.servers", "hadoop103:9092"); // 等待所有副本節點的應答 props.put("acks", "all"); // 消息發送最大嘗試次數 props.put("retries", 0); // 一批消息處理大小 props.put("batch.size", 16384); // 增加服務端請求延時 props.put("linger.ms", 1); // 發送緩存區內存大小 props.put("buffer.memory", 33554432); // key序列化 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
for (int i = 0; i < 50; i++) {
kafkaProducer.send(new ProducerRecord<String, String>("first", "hello" + i), new Callback() {
@Override public void onCompletion(RecordMetadata metadata, Exception exception) {
if (metadata != null) {
System.err.println(metadata.partition() + "---" + metadata.offset()); } } }); }
kafkaProducer.close(); } } |
4.2.3 自定義分區生產者
0）需求：將所有數據存儲到topic的第0號分區上
1）定義一個類實現Partitioner接口，重寫里面的方法（過時API）
| package com.atguigu.kafka; import java.util.Map; import kafka.producer.Partitioner;
public class CustomPartitioner implements Partitioner {
public CustomPartitioner() { super(); }
@Override public int partition(Object key, int numPartitions) { // 控制分區 return 0; } } |
2）自定義分區（新API）
| package com.atguigu.kafka; import java.util.Map; import org.apache.kafka.clients.producer.Partitioner; import org.apache.kafka.common.Cluster;
public class CustomPartitioner implements Partitioner {
@Override public void configure(Map<String, ?> configs) {
}
@Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 控制分區 return 0; }
@Override public void close() {
} } |
3）在代碼中調用
| package com.atguigu.kafka; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord;
public class PartitionerProducer {
public static void main(String[] args) {
Properties props = new Properties(); // Kafka服務端的主機名和端口號 props.put("bootstrap.servers", "hadoop103:9092"); // 等待所有副本節點的應答 props.put("acks", "all"); // 消息發送最大嘗試次數 props.put("retries", 0); // 一批消息處理大小 props.put("batch.size", 16384); // 增加服務端請求延時 props.put("linger.ms", 1); // 發送緩存區內存大小 props.put("buffer.memory", 33554432); // key序列化 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value序列化 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 自定義分區 props.put("partitioner.class", "com.atguigu.kafka.CustomPartitioner");
Producer<String, String> producer = new KafkaProducer<>(props); producer.send(new ProducerRecord<String, String>("first", "1", "atguigu"));
producer.close(); } } |
4）測試
（1）在hadoop102上監控/opt/module/kafka/logs/目錄下first主題3個分區的log日志動態變化情況
[atguigu@hadoop102 first-0]$ tail -f 00000000000000000000.log
[atguigu@hadoop102 first-1]$ tail -f 00000000000000000000.log
[atguigu@hadoop102 first-2]$ tail -f 00000000000000000000.log
（2）發現數據都存儲到指定的分區了。
4.3 Kafka消費者Java API
0）在控制台創建發送者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first
>hello world
1）創建消費者（過時API）
| package com.atguigu.kafka.consume; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; import kafka.consumer.Consumer; import kafka.consumer.ConsumerConfig; import kafka.consumer.ConsumerIterator; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector;
public class CustomConsumer {
@SuppressWarnings("deprecation") public static void main(String[] args) { Properties properties = new Properties();
properties.put("zookeeper.connect", "hadoop102:2181"); properties.put("group.id", "g1"); properties.put("zookeeper.session.timeout.ms", "500"); properties.put("zookeeper.sync.time.ms", "250"); properties.put("auto.commit.interval.ms", "1000");
// 創建消費者連接器 ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
HashMap<String, Integer> topicCount = new HashMap<>(); topicCount.put("first", 1);
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCount);
KafkaStream<byte[], byte[]> stream = consumerMap.get("first").get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) { System.out.println(new String(it.next().message())); } } } |
2）官方提供案例（自動維護消費情況）（新API）
| package com.atguigu.kafka.consume; import java.util.Arrays; import java.util.Properties; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer;
public class CustomNewConsumer {
public static void main(String[] args) {
Properties props = new Properties(); // 定義kakfa 服務的地址，不需要將所有broker指定上 props.put("bootstrap.servers", "hadoop102:9092"); // 制定consumer group props.put("group.id", "test"); // 是否自動確認offset props.put("enable.auto.commit", "true"); // 自動確認offset的時間間隔 props.put("auto.commit.interval.ms", "1000"); // key的序列化類 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // value的序列化類 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 定義consumer KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 消費者訂閱的topic, 可同時訂閱多個 consumer.subscribe(Arrays.asList("first", "second","third"));
while (true) { // 讀取數據，讀取超時時間為100ms ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } } } |
五 Kafka producer攔截器(interceptor)
5.1 攔截器原理
Producer攔截器(interceptor)是在Kafka 0.10版本被引入的，主要用於實現clients端的定制化控制邏輯。
對於producer而言，interceptor使得用戶在消息發送前以及producer回調邏輯前有機會對消息做一些定制化需求，比如修改消息等。同時，producer允許用戶指定多個interceptor按序作用於同一條消息從而形成一個攔截鏈(interceptor chain)。Intercetpor的實現接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定義的方法包括：
（1）configure(configs)
獲取配置信息和初始化數據時調用。
（2）onSend(ProducerRecord)：
該方法封裝進KafkaProducer.send方法中，即它運行在用戶主線程中。Producer確保在消息被序列化以計算分區前調用該方法。用戶可以在該方法中對消息做任何操作，但最好保證不要修改消息所屬的topic和分區，否則會影響目標分區的計算
（3）onAcknowledgement(RecordMetadata, Exception)：
該方法會在消息被應答之前或消息發送失敗時調用，並且通常都是在producer回調邏輯觸發之前。onAcknowledgement運行在producer的IO線程中，因此不要在該方法中放入很重的邏輯，否則會拖慢producer的消息發送效率
（4）close：
關閉interceptor，主要用於執行一些資源清理工作
如前所述，interceptor可能被運行在多個線程中，因此在具體實現時用戶需要自行確保線程安全。另外倘若指定了多個interceptor，則producer將按照指定順序調用它們，並僅僅是捕獲每個interceptor可能拋出的異常記錄到錯誤日志中而非在向上傳遞。這在使用過程中要特別留意。
5.2 攔截器案例
1）需求：
實現一個簡單的雙interceptor組成的攔截鏈。第一個interceptor會在消息發送前將時間戳信息加到消息value的最前部；第二個interceptor會在消息發送后更新成功發送消息數或失敗發送消息數。
2）案例實操
（1）增加時間戳攔截器
| package com.atguigu.kafka.interceptor; import java.util.Map; import org.apache.kafka.clients.producer.ProducerInterceptor; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata;
public class TimeInterceptor implements ProducerInterceptor<String, String> {
@Override public void configure(Map<String, ?> configs) {
}
@Override public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { // 創建一個新的record，把時間戳寫入消息體的最前部 return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(), System.currentTimeMillis() + "," + record.value().toString()); }
@Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
@Override public void close() {
} } |
（2）統計發送消息成功和發送失敗消息數，並在producer關閉時打印這兩個計數器
| package com.atguigu.kafka.interceptor; import java.util.Map; import org.apache.kafka.clients.producer.ProducerInterceptor; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata;
public class CounterInterceptor implements ProducerInterceptor<String, String>{ private int errorCounter = 0; private int successCounter = 0;
@Override public void configure(Map<String, ?> configs) {
}
@Override public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) { return record; }
@Override public void onAcknowledgement(RecordMetadata metadata, Exception exception) { // 統計成功和失敗的次數 if (exception == null) { successCounter++; } else { errorCounter++; } }
@Override public void close() { // 保存結果 System.out.println("Successful sent: " + successCounter); System.out.println("Failed sent: " + errorCounter); } } |
（3）producer主程序
| package com.atguigu.kafka.interceptor; import java.util.ArrayList; import java.util.List; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord;
public class InterceptorProducer {
public static void main(String[] args) throws Exception { // 1 設置配置信息 Properties props = new Properties(); props.put("bootstrap.servers", "hadoop102:9092"); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 2 構建攔截鏈 List<String> interceptors = new ArrayList<>(); interceptors.add("com.atguigu.kafka.interceptor.TimeInterceptor"); interceptors.add("com.atguigu.kafka.interceptor.CounterInterceptor"); props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
String topic = "first"; Producer<String, String> producer = new KafkaProducer<>(props);
// 3 發送消息 for (int i = 0; i < 10; i++) {
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message" + i); producer.send(record); }
// 4 一定要關閉producer，這樣才會調用interceptor的close方法 producer.close(); } } |
3）測試
（1）在kafka上啟動消費者，然后運行客戶端java程序。
[atguigu@hadoop102 kafka]$ in/kafka-console-consumer.sh --zookeeper hadoop102:2181 --from-beginning --topic first
1501904047034,message0
1501904047225,message1
1501904047230,message2
1501904047234,message3
1501904047236,message4
1501904047240,message5
1501904047243,message6
1501904047246,message7
1501904047249,message8
1501904047252,message9
（2）觀察java平台控制台輸出數據如下：
Successful sent: 10
Failed sent: 0
六 kafka Streams
6.1 概述
6.1.1 Kafka Streams
Kafka Streams。Apache Kafka開源項目的一個組成部分。是一個功能強大，易於使用的庫。用於在Kafka上構建高可分布式、拓展性，容錯的應用程序。
6.1.2 Kafka Streams特點
1）功能強大
高擴展性，彈性，容錯
2）輕量級
無需專門的集群
一個庫，而不是框架
3）完全集成
100%的Kafka 0.10.0版本兼容
易於集成到現有的應用程序
4）實時性
毫秒級延遲
並非微批處理
窗口允許亂序數據
允許遲到數據
6.1.3 為什么要有Kafka Stream
當前已經有非常多的流式處理系統，最知名且應用最多的開源流式處理系統有Spark Streaming和Apache Storm。Apache Storm發展多年，應用廣泛，提供記錄級別的處理能力，當前也支持SQL on Stream。而Spark Streaming基於Apache Spark，可以非常方便與圖計算，SQL處理等集成，功能強大，對於熟悉其它Spark應用開發的用戶而言使用門檻低。另外，目前主流的Hadoop發行版，如Cloudera和Hortonworks，都集成了Apache Storm和Apache Spark，使得部署更容易。
既然Apache Spark與Apache Storm擁用如此多的優勢，那為何還需要Kafka Stream呢？筆者認為主要有如下原因。
第一，Spark和Storm都是流式處理框架，而Kafka Stream提供的是一個基於Kafka的流式處理類庫。框架要求開發者按照特定的方式去開發邏輯部分，供框架調用。開發者很難了解框架的具體運行方式，從而使得調試成本高，並且使用受限。而Kafka Stream作為流式處理類庫，直接提供具體的類給開發者調用，整個應用的運行方式主要由開發者控制，方便使用和調試。
第二，雖然Cloudera與Hortonworks方便了Storm和Spark的部署，但是這些框架的部署仍然相對復雜。而Kafka Stream作為類庫，可以非常方便的嵌入應用程序中，它對應用的打包和部署基本沒有任何要求。
第三，就流式處理系統而言，基本都支持Kafka作為數據源。例如Storm具有專門的kafka-spout，而Spark也提供專門的spark-streaming-kafka模塊。事實上，Kafka基本上是主流的流式處理系統的標准數據源。換言之，大部分流式系統中都已部署了Kafka，此時使用Kafka Stream的成本非常低。
第四，使用Storm或Spark Streaming時，需要為框架本身的進程預留資源，如Storm的supervisor和Spark on YARN的node manager。即使對於應用實例而言，框架本身也會占用部分資源，如Spark Streaming需要為shuffle和storage預留內存。但是Kafka作為類庫不占用系統資源。
第五，由於Kafka本身提供數據持久化，因此Kafka Stream提供滾動部署和滾動升級以及重新計算的能力。
第六，由於Kafka Consumer Rebalance機制，Kafka Stream可以在線動態調整並行度。
6.2 Kafka Stream案例
6.2.1 eclipse打包插件安裝
1）將net.sf.fjep.fatjar_0.0.32.jar拷貝到eclipse安裝目錄中的plugins目錄下，然后重啟eclipse即可。
2）插件使用方法
6.2.2 數據清洗案例
0）需求：
實時處理單詞帶有”>>>”前綴的內容。例如輸入”atguigu>>>ximenqing”，最終處理成“ximenqing”
1）創建主類
| package com.atguigu.kafka.stream; import java.util.Properties; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorSupplier; import org.apache.kafka.streams.processor.TopologyBuilder;
public class Application {
public static void main(String[] args) {
// 定義輸入的topic String from = "first"; // 定義輸出的topic String to = "second";
// 設置參數 Properties settings = new Properties(); settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "logFilter"); settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
StreamsConfig config = new StreamsConfig(settings);
// 構建拓撲 TopologyBuilder builder = new TopologyBuilder();
builder.addSource("SOURCE", from) .addProcessor("PROCESS", new ProcessorSupplier<byte[], byte[]>() {
@Override public Processor<byte[], byte[]> get() { // 具體分析處理 return new LogProcessor(); } }, "SOURCE") .addSink("SINK", to, "PROCESS");
// 創建kafka stream KafkaStreams streams = new KafkaStreams(builder, config); streams.start(); } } |
2）具體業務處理
| package com.atguigu.kafka.stream; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext;
public class LogProcessor implements Processor<byte[], byte[]> {
private ProcessorContext context;
@Override public void init(ProcessorContext context) { this.context = context; }
@Override public void process(byte[] key, byte[] value) { String input = new String(value);
// 如果包含“>>>”則只保留該標記后面的內容 if (input.contains(">>>")) { input = input.split(">>>")[1].trim(); // 輸出到下一個topic context.forward("logProcessor".getBytes(), input.getBytes()); }else{ context.forward("logProcessor".getBytes(), input.getBytes()); } }
@Override public void punctuate(long timestamp) {
}
@Override public void close() {
} } |
3）將程序用eclipse插件打成jar包
4）將jar包拷貝hadoop102上運行
[atguigu@hadoop102 kafka]$ java -jar kafka0508_fat.jar com.atguigu.kafka.stream.Application
5）在hadoop104上啟動生產者
[atguigu@hadoop104 kafka]$ bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first
>hello>>>world
>h>>>atguigu
>hahaha
6）在hadoop103山啟動消費者
[atguigu@hadoop103 kafka]$ bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --from-beginning --topic second
world
atguigu
hahaha
說明：
本文來源：尚硅谷大數據技術之Kafka （作者：大海哥）