一、什么是EFAK
EFAK（Eagle For Apache Kafka，以前稱為 Kafka Eagle）EFAK是開源可視化和管理軟件?？梢圆樵?、可視化、監(jiān)控kafka集群，是將 kafka 的集群數據轉換為圖形可視化的工具。
二、為什么要用EFAK
Apache Kafka 沒有正式提供監(jiān)控系統(tǒng)或頁面。
開源Kafka監(jiān)控系統(tǒng)功能太少或暫停維護。
現(xiàn)有的監(jiān)控系統(tǒng)難以配置和使用。
一些監(jiān)控系統(tǒng)無法滿足與現(xiàn)有IM的集成，如微信、釘釘等。
三 、安裝
3.1 下載
可以在 GitHub 上下載EFAK源代碼自行編譯安裝，也可以下載二進制 .tar.gz 文件。
ps：推薦使用官方編譯的二進制安裝包
3.2 安裝JDK
如果Linux服務器上有JDK環(huán)境，這一步可以忽略，繼續(xù)下一步的安裝。如果沒有JDK，請先到Oracle官網下載JDK。
JAVA_HOME 配置
將二進制安裝包解壓到指定目錄：
cd /usr/java
tar -zxvf jdk-xxxx.tar.gz
mv jdk-xxxx jdk1.8
vi /etc/profile
export JAVA_HOME=/usr/java/jdk1.8
export PATH=$PATH:$JAVA_HOME/bin
然后，我們使用. /etc/profile使配置立即生效。
3.3 提取 EFAK
這里我們解壓到/data/soft/new目錄并解壓
tar -zxvf efak-xxx-bin.tar.gz
如果之前安裝過版本，刪除修改后的版本，重命名當前版本
rm -rf efakmv efak-xxx efak
然后，配置 EFAK 配置文件
vi /etc/profileexport KE_HOME=/data/soft/new/efakexport PATH=$PATH:$KE_HOME/bin
最后，我們使用. /etc/profile使配置立即生效。
配置EFAK系統(tǒng)文件:
根據自身Kafka集群的實際情況配置EFAK，例如zookeeper地址、Kafka集群的版本類型（zk為低版本，kafka為高版本）、開啟安全認證的Kafka集群等。
cd ${KE_HOME}/conf
vi system-config.properties
# Multi zookeeper&kafka cluster list -- The client connection address of the Zookeeper cluster is set here
efak.zk.cluster.alias=cluster1,cluster2
cluster1.zk.list=tdn1:2181,tdn2:2181,tdn3:2181
cluster2.zk.list=xdn1:2181,xdn2:2181,xdn3:2181
# Add zookeeper acl
cluster1.zk.acl.enable=false
cluster1.zk.acl.schema=digest
cluster1.zk.acl.username=test
cluster1.zk.acl.password=test123
# Kafka broker nodes online list
cluster1.efak.broker.size=10
cluster2.efak.broker.size=20
# Zkcli limit -- Zookeeper cluster allows the number of clients to connect to
# If you enable distributed mode, you can set value to 4 or 8
kafka.zk.limit.size=16
# EFAK webui port -- WebConsole port access address
efak.webui.port=8048
######################################
# EFAK enable distributed
######################################
efak.distributed.enable=false
# master worknode set status to master, other node set status to slave
efak.cluster.mode.status=slave
# deploy efak server address
efak.worknode.master.host=localhost
efak.worknode.port=8085
# Kafka offset storage -- Offset stored in a Kafka cluster, if stored in the zookeeper, you can not use this option
cluster1.efak.offset.storage=kafka
cluster2.efak.offset.storage=kafka
# Whether the Kafka performance monitoring diagram is enabled
efak.metrics.charts=false
# EFAK keeps data for 30 days by default
efak.metrics.retain=30
# If offset is out of range occurs, enable this property -- Only suitable for kafka sql
efak.sql.fix.error=false
efak.sql.topic.records.max=5000
# Delete kafka topic token -- Set to delete the topic token, so that administrators can have the right to delete
efak.topic.token=keadmin
# Kafka sasl authenticate
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin-secret";
# If not set, the value can be empty
cluster1.efak.sasl.client.id=
# Add kafka cluster cgroups
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=kafka_ads01,kafka_ads02
cluster2.efak.sasl.enable=true
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";
cluster2.efak.sasl.client.id=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=kafka_ads03,kafka_ads04
# Default use sqlite to store data
efak.driver=org.sqlite.JDBC
# It is important to note that the '/hadoop/kafka-eagle/db' path must be exist.
efak.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
efak.username=root
efak.password=smartloli
# (Optional) set mysql address
#efak.driver=com.mysql.jdbc.Driver
#efak.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
#efak.username=root
#efak.password=smartloli
3.4 啟動EFAK服務器（獨立）
在$KE_HOME/bin目錄中，有一個ke.sh腳本文件
cd ${KE_HOME}/binchmod +x ke.sh ke.sh start
之后，當 EFAK 服務器重新啟動或停止時，執(zhí)行以下命令
ke.sh restartke.sh stop
3.5 啟動EFAK服務器（分布式）
在$KE_HOME/bin目錄中，有一個ke.sh腳本文件。執(zhí)行啟動命令如下
cd ${KE_HOME}/bin
# sync efak package to other worknode node
# if $KE_HOME is /data/soft/new/efak
for i in `cat $KE_HOME/conf/works`;do scp -r $KE_HOME $i:/data/soft/new;done
# sync efak server .bash_profile environment
for i in `cat $KE_HOME/conf/works`;do scp -r ~/.bash_profile $i:~/;done
chmod +x ke.sh
ke.sh cluster start
之后，當 EFAK 服務器重新啟動或停止時，執(zhí)行以下命令
ke.sh cluster restart
ke.sh cluster stop
四、使用
4.1 儀表盤
查看Kafka brokers、topics、consumer 、Zookeepers信息等
1
4.2 創(chuàng)建主題
2
4.3 列出主題
該模塊跟蹤Kafka集群中的所有Topic，包括分區(qū)數、創(chuàng)建時間、修改Topic，如下圖所示：
3
4.4 主題詳情
4
4.5 消費情況
5
4.6 數據預警
配置下郵件服務器的用戶名和密碼，就可以看到相應的預警數據了
6
4.7 以及最后的大屏展示
7
五、補充
推薦Kafka另外一個可視化管理工具kafka-ui，代碼優(yōu)雅但是功能沒這個強
