0、说明本篇文章，有些片段和图片摘自网络，因为杂碎，就不列出取自的地址，在此先表达对他们的感谢

1、部署规划

1. elasticsearch-v7.3.2 10.201.5.215/216/217

2. logstash-v7.3.2 10.201.5.215/216

3. kibana-v7.3.2 10.201.5.217

4. kafka+zookeeper-v2.12-2.3.0 10.201.5.215/216/217

5. filebeat-v7.3.2：10.201.5.215/216

6. 数据流向如下图：

2、准备

所有机器都关闭防火墙和selinux

systemctl stop firewalld
systemctl disable firewalld
vim /etc/sysconfig/selinux
SELinux=disabled
setenforce 0

3、安装elasticsearch

1.）设置文件打开数限制

vim /etc/security/limits.conf

* soft nofile 65536
* hard nofile 65536

vim /etc/sysctl.conf

vm.max_map_count = 555555 #这个值至少要大于262144 不然启动会报错

sysctl -p
systemctl reboot -i

2.）下载elasticsearch源码包

cd /usr/local/src
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.3.2-linux-x86_64.tar.gz

3.) 解压配置并启动

解压：

cd /usr/local/src && tar -xzvf  elasticsearch-7.3.2-linux-x86_64.tar.gz -C /usr/local

因为es7.2及以上版本要求java11及以上的安装环境，从官网查询得知，es自带了java11，所以我们只要把我们本地的java环境变量给注释掉重启即可。如果该机器上确实又有其他项目需要用到java8怎么办，那么就直接修改es配置让它使用自己自带的java11。
vim bin/elasticsearch 需要在前面加这javahome目录这一行。

JAVA_HOME='/usr/local/elasticsearch-7.3.2/jdk'
source "`dirname "$0"`"/elasticsearch-env

配置：
cd /usr/local/elasticsearch-7.3.2/config
vim elasticsearch.yml 需要配置下面这些项，具体每个什么意思，请自行百度，不再标出

cluster.initial_master_nodes:
- es_cluster_node01
- es_cluster_node02

cluster.name: cyk_es_cluster
node.name: es_cluster_node03
node.master: false
node.data: true
path.data: /data/elasticsearch/data
path.logs: /data/elasticsearch/logs
bootstrap.memory_lock: false
bootstrap.system_call_filter: false
network.host: 0.0.0.0
http.port: 9200
transport.profiles.default.port: 9300
discovery.seed_hosts: ["es_cluster_node03","es_cluster_node02","es_cluster_node01"]
#discovery.zen.minimum_master_nodes: 2
discovery.zen.ping_timeout: 150s
discovery.zen.fd.ping_retries: 10
client.transport.ping_timeout: 60s
http.cors.enabled: true
http.cors.allow-origin: “*”
logger.org.elasticsearch.cluster.coordination: TRACE

创建日志和数据目录
mkdir -p /data/elasticsearch/{data,log}

启动：

cd /usr/local/elasticsearch-7.3.2/bin;./elasticsearch -d

因为我是以root执行的，所以启动遇到报错：java.lang.RuntimeException: can not run elasticsearch as root。
新建一个用户elastic用户,同时把安装目录和存放日志及数据的目录也更改一下属主，使得elastic用户有读写权限

useradd elastic -g root
chown elastic:root -R  /usr/local/elasticsearch-7.3.2
chown elastic:root -R  /data/elasticsearch/

切换到elastic用户继续启动：

su - elastic
./elasticsearch -d

验证：curl 127.0.0.1:9200 和 curl http://10.201.5.213:9200/_cat/master
有如下信息返回就说明安装ok

{
"name" : "es_cluster_node03",
"cluster_name" : "cyk_es_cluster",
"cluster_uuid" : "-MCxwBJ-Ra6tuMeU7Z94ZA",
"version" : {
"number" : "7.3.2",
"build_flavor" : "default",
"build_type" : "tar",
"build_hash" : "1c1faf1",
"build_date" : "2019-09-06T14:40:30.409026Z",
"build_snapshot" : false,
"lucene_version" : "8.1.0",
"minimum_wire_compatibility_version" : "6.8.0",
"minimum_index_compatibility_version" : "6.0.0-beta1"
},
"tagline" : "You Know, for Search"
}
和
aNPEDiapRBm1CfirJW2U4w 10.201.5.216 10.201.5.216 es_cluster_node02   #尤其这条显示了集群选出了主

4、安装kibana

4.1 下载kibana并解压

cd /usr/local/src
wget https://artifacts.elastic.co/downloads/kibana/kibana-7.3.2-linux-x86_64.tar.gz
tar -xzvf kibana-7.3.2-linux-x86_64.tar.gz -C /usr/local/

4.2 修改配置

cd /usr/local/kibana-7.3.2
vim config/kibana.yml
注意： elasticsearch的地址要写上面我们安装好的elasticsearch的地址

server.port: 5601
server.host: "10.201.5.217"
elasticsearch.hosts: ["http://10.201.5.215:9200"]

4.3 启动kibana

nohup ./bin/kibana &
启动报错：同样也是不能以root启动，解决办法和上面一样。加个elastic用户，然后以elastic用户来启动
然后浏览器访问： 10.201.5.217:5601。

5、安装并配置kafka+zookeeper

安装了kafka以后，kafka的包里面就自带了zookeeper，所以不需要额外的安装
下载解压两者的集成包

cd /usr/local/src
wget http://apache-mirror.8birdsvideo.com/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar -xzvf kafka_2.12-2.3.0.tgz  -C /usr/local

5.1 安装配置zk

5.1.1 配置zk

mkdir -p /data/zookeeper/{data,logs}
vim /usr/local/kafka_2.12-2.3.0/config/zookeeper.properties
对下面关键部分信息说明：
说明：server.A=B:C:D 其中A是一个数字，表示这个是第几号服务器；B是这个服务器的IP地址（如果做了域名映射，也可以用域名）；C表示的是这个服务器与集群中的Leader服务器交换信息的端口；D表示的是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

dataDir=/data/zookeeper/data
dataLogDir=/data/zookeeper/logs
clientPort=2181
tickTime=2000
initLimit=20
syncLimit=10
server.215=10.201.5.215:2888:3888
server.216=10.201.5.216:2888:3888
server.217=10.201.5.217:2888:3888

特别说明一下，下面配置的myid必须和上面zk配置的编号对应（即：server.215=10.201.5.215:2888:3888中的215这样的）
在10.201.5.215的/data/zookeeper/data目录下面执行：echo 215 > myid
在10.201.5.216的/data/zookeeper/data目录下面执行：echo 216 > myid
在10.201.5.217的/data/zookeeper/data目录下面执行：echo 217 > myid

5.1.2 启动zk

启动之前，先把zk的数据和日志目录改一下属主：chown elastic:root -R /data/zookeeper
nohup /usr/local/kafka_2.12-2.3.0/bin/zookeeper-server-start.sh /usr/local/kafka_2.12-2.3.0/config/zookeeper.properties > /data/zookeeper/logs/zk_output.log 2>&1 &

5.1.3 查看zk集群信息

抱歉，使用kafka自带的zk，我还不知道怎么查看集群的信息，如果有知道的大佬，可以评论告知我一下。一个土的验证办法就是在某一台机器上创建一个topic，然后去其他机器上，查看一下topic信息是否写入到了zk中。如果能写入，那就说明集群是没有问题的。但是至于集群中，节点角色，就没有办法知道了，我也不知道三个节点中哪个是leader，也没有相应的命令查看。还有一个确定zk集群是否正常的一点是，在zk存储中的cluster下面看一下id是否和其他节点一样，如果都是一样，说明集群是ok的。简单贴两句命令和截图【执行下面这两句话，还要把kafka也运行起来才行】

/usr/local/kafka_2.12-2.3.0/bin/kafka-topics.sh --create --zookeeper 10.201.5.217:2181,10.201.5.216:2181,10.201.5.215:2181 --replication-factor 1 --partitions 1 --topic testtopic/usr/local/kafka_2.12-2.3.0/bin/kafka-topics.sh --zookeeper 10.201.5.217:2181,10.201.5.216:2181,10.201.5.215:2181 --list

5.1.4 kafka的ZooKeeper存储结构

5.2 配置启动kafka

5.2.1 配置kafka

下面就只展示一个节点下安装的两个broker的配置。其他两个节点也是类似的。只要改几个必要变的地方就ok了
cat server-1.properties |grep -vE “#|$”

broker.id=1
listeners=PLAINTEXT://10.201.5.217:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs-service-1
num.partitions=2
num.recovery.threads.per.data.dir=3
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.201.5.215:2181,10.201.5.216:2181,10.201.5.217:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true

cat server-2.properties |grep -v ^# |grep -v ^$

broker.id=2
listeners=PLAINTEXT://10.201.5.217:9093
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs-server-2
num.partitions=3
num.recovery.threads.per.data.dir=3
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.201.5.215:2181,10.201.5.216:2181,10.201.5.217:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true

5.2.2 启动kafka

老规矩启动之前，先把kafka的数据和日志目录改一下属主：chown elastic:root -R /data/kafka
nohup /usr/local/kafka_2.12-2.3.0/bin/kafka-server-start.sh /usr/local/kafka_2.12-2.3.0/config/server-1.properties > /data/kafka/logs/run_output.log 2>&1 &

5.2.3 模拟消息生产和消费

bin/kafka-console-producer.sh --broker-list 10.201.5.217:9092 --topic cyk_filebeat_config_output.kafka
bin/kafka-console-consumer.sh --bootstrap-server 192.168.88.129:9092 --topic testtopic --from-beginning

如果要查看某个kafka中的消息：

生产者发送消息

消费者会在另外一端同步的接收到消息

5.2.4 其他语法

查看当前broker上的消费组:

./kafka-consumer-groups.sh --bootstrap-server 10.202.1.11:9092 --list --new-consumer

查看某个消费组的消息堆积情况：

./kafka-consumer-groups.sh --bootstrap-server 10.202.1.11:9092 --describe --group xinyada

创建topic：

/usr/local/kafka_2.12-2.3.0/bin/kafka-topics.sh --create --zookeeper 10.201.5.217:2181,10.201.5.216:2181,10.201.5.215:2181 --replication-factor 1 --partitions 1 --topic testtopic

查看所有topic：

/usr/local/kafka_2.12-2.3.0/bin/kafka-topics.sh --list --zookeeper 10.201.5.217:2181,10.201.5.216:2181,10.201.5.215:2181

删除某个topic(也不知道是那里的问题，我删除topic，需要先把logstash停掉，猜想可能是logstash继续拉数据的时候，打开了该topic下某些文件的文件描述符，故需要先停止logstash关闭连接，才能继续删除，真正原因是logstash会反写数据到kafka，没有明白为啥能反写，经过试验得知，logstash会根据自己的配置文件中的input下面的topics的值到kafka集群中写入tipic信息，当然filebeat也能写入topic信息。要想真正删除一个tipic，就两步，第一步logstash配置文件里面要把这个要删除的topic字段配置给删除掉，然后重启logstash，第二步用下面命令再到kafka集群中删除topic,后面自己又做测试，发现可能是因为zk的数据目录里面还存在，或者说kafka的日志目录里面有，所以删除到下个时间间隔后，又去读日志目录发现还存在，就又写入了。所以我直接把kafka日志目录和zk的数据目录给删除了，就ok了)：

/usr/local/kafka_2.12-2.3.0/bin/kafka-topics.sh --delete --zookeeper 10.201.5.217:2181,10.201.5.216:2181,10.201.5.215:2181 --topic cyk_filebeat_config_output.kafka

6. 安装filebeat搜集日志

6.1 下载filebeat并解压

cd  /usr/local/src
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.3.2-linux-x86_64.tar.gz
tar -xzvf filebeat-7.3.2-linux-x86_64.tar.gz -C /usr/local
cd /usr/local/  &&  mv filebeat-7.3.2-linux-x86_64/ filebeat-7.3.2

6.2 配置filebeat

先把filebeat的目录更改一下属主 chown elastic:root -R /usr/local/filebeat-7.3.2
vim cyk_filebeat.yml

filebeat.inputs:
- type: log   #这里注意要用type不要用input_type。不然可能接收到数据。
tail_files: true
paths:
- /var/log/cyk215.log
fields:
appid: appid_cyk_filebeat_config_tag_input_cyk215

output.kafka:
hosts: ["10.201.5.215:9092","10.201.5.215:9093","10.201.5.216:9092","10.201.5.216:9093","10.201.5.217:9092","10.201.5.217:9093"]
topic: 'topic_cyk_filebeat_config_output.kafka_cyk215'
partition.round_robin:
reachable_only: false

required_acks: 1
compression: gzip
max_message_bytes: 1000000

6.3启动filebeat

nohup /usr/local/filebeat-7.3.2/filebeat -c /usr/local/filebeat-7.3.2/cyk_filebeat.yml &

7. 安装logstash

7.1 下载logstash并解压

cd /usr/local/src
wget  https://artifacts.elastic.co/downloads/logstash/logstash-7.3.2.tar.gz
tar -xzvf logstash/logstash-7.3.2.tar.gz -C /usr/local

7.2 编辑配置文件

先把logstash的目录更改一下属主 chown elastic:root -R /usr/local/logstash-7.3.2
vim /usr/local/logstash-7.3.2/config/cyk_logstash.conf 添加如下内容

input {
kafka {
bootstrap_servers => ["10.201.5.217:9092,10.201.5.217:9093,10.201.5.216:9092,10.201.5.216:9093,10.201.5.215:9092,10.201.5.215:9093"]
topics => "topic_cyk_filebeat_config_output.kafka_cyk215"
codec => "json"
}
}

output {
elasticsearch {
hosts => ["10.201.5.215:9200","10.201.5.216:9200","10.201.5.217:9200"]
index => "%{[fields][appid]}-%{+YYYY.MM.dd}"
codec => "json"
}
}

7.3 启动logstash

nohup /usr/local/logstash-7.3.2/bin/logstash -f /usr/local/logstash-7.3.2/config/cyk_logstash.conf &

最终效果

7.4 其他

如果有时，使用es的api查看index时比如：http://10.201.5.213:9200/_cat/indices，没有查看到自己新加的，可能需要重新往监控的日志文件里面写数据，不然index不会刷新。

• 点赞
• 收藏
• 分享
• 文章举报

~银开~ 发布了19 篇原创文章 · 获赞 0 · 访问量 890 私信 关注