zk+kafka+elk

Posted 2020-11-03 gzcheng

软件版本
zookeeper 3.4.10
kafka 2.11-0.10.2.0
elaticsearch 2.10.16
logstash 2.10.2
kibana 4.3.1
jdk 1.8.0_171
服务器准备：
10.253.2.50
10.253.2.51
10.553.2.41
本次安装均在linux系统下操作
软件均安装在 /usr/local/share/applications/目录下
日志输出均在 /tmp/包名
一软件安装及配置
1.jdk安装
检测服务器的Java版本 Java -version

本机Java已经满足要求 不在赘述
2.zk安装
到官网下载相关软件包
sudo tar -xvf zookeeper-3.4.10.tar.gz -C /usr/local/share/applications/
修改配置文件
进入conf目录下 cp zoo_sample.cfg zoo.cfg
tickTime ZooKeeper 中 的一个时间单元 。 ZooKeepe r 中所有时间都以这
个时间单元为基准 ， 进行整数倍配置 ， 默认是 2s
initLimit Follower 在启动过程中 ， 会从 Leader 同步所有最新数据 ， 确定自
己能够对外服务的起始状态。当 Follower 在 initLimt 个 tickTime
还没有完成数据同步时，则 Leader 认为 Follower 连接失败
syncLimit Leader 与 Follower 之间通信请求和应答的时间长度 。若 Leader
在 syncLimit 个 tickTime 还没有收到 Follower 应答，则认为该
Leader 己下线
dataDir 存储快照文件的 目录，默认情况下 ， 事务日志也会存储在该目
录上。由于事务日志的写性能直接影响 ZooKeeper 性能，因此
建议同时配置参数 dataLogDir
clientPort ZooKeeper 对外端口
dataDir 事务日志输出目录
集群信息的配置（首先需要在/etc/hosts下面配置解析）
server.1=10.253.2.50:2888:3888
server.2=10.253.2.51:2888:3888
server.3=10.253.2.41:2888:3888
（2888是服务器与leader通信的端口 3888是选举时相互通信的端口）
Kafka 依赖 ZooKeeper ，通过 ZooKeeper 来对代理、消费者上下线管理、集群、分区元数
据管理等，因此 ZooKeeper 也是 Kafka 得以运行的基础环境之一。
关于zk之于kafka的具体内容 可以参考zk官网
在zk bin目录下 ./zkServer.sh start/status

3.kafka安装
sudo tar -xvf kafka_2.11-0.10.2.0.tgz -C /usr/local/share/applications/
修改配置文件
进入config文件
vi server.properties
broker.id=1
lod.dirs=/tmp/kafak
zookeeper.connect=10.253.2.50:2181,10.253.2.51:2181,10.253.2.41:2181
hostname=ip
启动
cd ../bin
kafka-server-start . sh - daemon . . /config/server .properties
注：2 3操作分别在51 41上面操作
进入zk bin下面
./zkCli.sh -server 10.253.2.51:2181

4.elk安装及配置
elasticsearch的搭建
1.可以选择tar包安装或者rpm包安装 殊途同归 这里选择rpm安装
rpm -ivh elsticsearch.6.1.2.rpm
2.修改配置文件 以及权限的变更
chown -R patrol:patrol /usr/share/elasticsearch
chown -R patrol:patrol /etc/elasticsearch/elasticsearch.yml
chown -R patrol:patrol /etc/sysconfig/elasticsearch

配置文件主要包括 elasticsearch.yml 和 sysctl.conf (需要执行sysctl -p 生效)
vi /etc/elasticsearch/elasticsearch.yml
cluster.name: ELK_node_0
node.name: ELK_node_0_4
path.data: /data/data_sdb/elasticsearch
path.logs: /data/data_sdb/log
bootstrap.memory_lock: false
bootstrap.system_call_filter : false
network.host: 10.253.2.50
transport.tcp.port: 9301

vi /etc/sysctl.conf
添加一行
vm.max_map_count = 524288

启动
/usr/share/elsticsearch/bin/elsticsearch -d

测试
curl ‘http://10.253.2.50:9201/?pretty’

logstach
简介：
Logstash是一个完全开源的工具，他可以对你的日志进行收集、过滤，并将其存储供以后使用（如，搜索）
安装
采用rpm包安装
rpm -ivh logstash-6.1.2.rpm

配置：
mkdir -p /data/data_sdb/logstash
mkdir -p /data/data_sdb/log/logstash
修改配置文件
vi /etc/logstash/conf.d/elasticsearch.conf
input {
stdin { }
}
output {
elasticsearch {
hosts => ["10.253.2.50:9201"]
}
stdout {
codec => rubydebug
}
}
vi /etc/logstash/logstash.yml
修改几行：
path.data: /data/data_sdb/logstash
path.logs: /data/data_sdb/log/logstash

chown -R patrol:patrol /usr/share/logstash
chown -R patrol:patrol /etc/logstash
启动：
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/elasticsearch.conf
kibana
简介：
Kibana 也是一个开源和免费的工具，Kibana可以为 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以帮助您汇总、分析和搜索重要数据日志
新建文件夹 装包
mkdir /data/data_sdb/log/kibana
rpm -ivh kibana-6.1.2-x86_64.rpm

chown -R patrol:patrol /etc/kibana
chown -R patrol:patrol /usr/share/kibana

修改配置文件

vi /etc/kibana/kibana.yml
server.port: 5601
server.host: "10.253.2.50"
elasticsearch.url: "http://10.253.2.50:9201"
pid.file: /var/run/kibana/kibana.pid

mkdir /var/run/kibana/
chown -R patorl:patrol /var/run/kibana

启动
/usr/share/kibana/bin/kibana

键值优化以及集群的建设

重启elaticsearch

curl 10.253.2.50:9201/_cat/health?v
查看节点健康状态

 

测试：
修改配置文件 /etc/logstash/conf.d/elasticsearch.conf
input {
file {
path => "/data/data_sdb/elasticsearch/*.log"
}
}

output {
elasticsearch { hosts => ["10.253.2.50:9201"] }
stdout {
codec => rubydebug
启动 logstash
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/elasticsearch.conf

当出现
WARNING: Could not find logstash.yml which is typically located in $LS_HOME/config or /etc/logstash. You can specify the path using --path.settings. Continuing using the defaults
Could not find log4j2 configuration at path /usr/share/logstash/config/log4j2.properties. Using default config which logs errors to the console
时 编辑
data/data_sdb/elasticsearch/*.log 文件

{
"@timestamp" => 2018-02-28T08:34:36.276Z,
"message" => "adsf",
"@version" => "1",
"path" => "/data/data_sdb/elasticsearch/1.log",
"host" => "centos-wz3hgym2"
}
{
"@timestamp" => 2018-02-28T08:34:36.274Z,
"message" => "fifdsfds",
"@version" => "1",
"path" => "/data/data_sdb/elasticsearch/1.log",
"host" => "centos-wz3hgym2"

}
你编辑的内容会以这样的形式出现
启动kibana 页面里面会出现相关的内容

 

elasticsearch配置文件参数修改

# 设置索引的分片数,默认为5
index.number_of_shards: 15

# 临时文件的路径
path.work: /path/to/work

# 插件安装路径
# path.plugins: /path/to/plugins

# 当JVM开始写入交换空间时（swapping）ElasticSearch性能会低下,你应该保证它不会写入交换空间
bootstrap.mlockall: true

# 探查的超时时间,默认3秒,提高一点以应对网络不好的时候,防止脑裂
discovery.zen.ping.timeout: 60s

二相关测试
kafka与elk的结合使用
首先 创建一个主题 进入kafka bin目录
kafka-topics.sh --zookeeper 10.253.2.50:2181,10.253.2.51:2181,10.253.2.41:2181 --create --topic kafka-elk-log --partitions 1 --replication-factor 1
主题名字为 kafka-elk-log
1.logstash传数据到kafka
进入logstash目录 mkdir etc
vi logstash-input-kafka
input{
＃从控制台接收输入
stdin{}
output{
kafka{
＃消息、写入的主题
top 工 c_id => ” kafka-elk-log ”
＃连接 Kafka 集群配置
bootstrap_servers => ” 10.253.2.50:2181,10.253.2.51:2181,10.253.2.41:2181 ”
＃批量写入配置
batch size => 5
# logstash 导入数据编码方式
codec => ” plain ”
stdout{
＃设置控制台打印数据表现形式
codec => rubydebug
}
}

cd bin/
logstash -f ../etc/logstash*
屏幕输出

验证：
kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/kafkaelk-log-0/00000000000000000000 . log --print-data-log

2.kafka数据传到es
vi logstash-output-es
input{
kafka {
# logstash 导出数据解码方式
codec => ” plain ”
＃消费组
group_id => ” kafka_elk_group”
＃消费者标识
cl 工 ent_id => logstash
＃消费的主题
topics =>”kafka_elk_log "
＃连接 Kafka 集群配置
bootstrap_servers => ”10.253.2.50:2181,10.253.2.51:2181,10.253.2.41:2181 ”
＃消费起始位置
auto offset reset => ” earliest ”
＃消费者：统程数
consumer threads => 5
output{
＃导入 ela sticsearch
elasticsearch {
# elasticsearch 集群地址， 多个地址以逗号隔开
hosts => ［ ” 10.253.2.50:9201”］
＃指定数据导入 elasticsearch 格式
codec => ” plain”
＃创建索 号｜
index => ” kaf ka_elk_log － ”｛＋ YYYY.MM .dd ｝ ”
}
}
验证：
浏览器：http://10.253.2.50:9201/_plugin/head/

kibana验证：

需要现在seting下面添加索引 索引名字在logstash配置文件中写好