赞
踩
相关中文文档:https://es.xiaoleilu.com 下载:从https://www.elastic.co/downloads/elasticsearch 下载对应操作系统版本的elasticsearch解压:tar -zxvf elasticsearch-5.2.0.tar.gz启动:./bin/elasticsearch.sh验证:elasticsearch --version默认端口:9200配置elasticsearch:打开config/elasticsearch.yml修改以下配置:
path.data: /Users/chenshunyang/tools/elasticsearch/data
path.logs: /Users/chenshunyang/tools/elasticsearch/logs
cluster.name: elktest //更改默认的名称为我们想要的名称
node.name: node-elktest //更改节点默认的名称
network.host: localhost
elasticsearch 2.x 版本
elasticsearch 2.x版本中,直接使用/bin/plugin install mobz/elasticsearch-head安装即可验证:通过访问 http://localhost:9200/_plugin/head/
elasticsearch 5.1 版本
安装:elasticsearch 5.1中,elasticsearch-head插件不能放在elasticsearch的 plugins、modules 目录下,不能使用 elasticsearch-plugin install,直接启动elasticsearch即可
git clone https://github.com/mobz/elasticsearch-head.git
cd elasticsearch-head/
npm install grunt
配置:修改elasticsearch/config/elasticsearch.yml 添加
http.cors.enabled: true
http.cors.allow-origin: "*"
打开elasticsearch-head-master/Gruntfile.js,找到下面connect属性,新增hostname: ‘0.0.0.0’:
connect: {
server: {
options: {
hostname: '0.0.0.0',
port: 9100,
base: '.',
keepalive: true
}
}
}
启动插件:
elasticsearch-head/ && ./node_modules/grunt/bin/grunt server
查看 http://localhost:9100
下载:从https://www.elastic.co/downloads/kibana 下载对应操作系统版本的kibana解压: tar -xzvf kibana-5.2.0-darwin-x86_64.tar.gz配置kibana:进入kibana/config打开kibana.yml,设置elasticsearch.url 为es实例的地址eg:elasticsearch.url: "http://localhost:9200"启动:./bin/kibana.sh默认端口:5601访问:http://localhost:5601
下载:从https://www.elastic.co/downloads/logstash 下载对应操作系统版本的logstash解压:tar -zxvf logstash-5.2.0.tar.gz验证:./bin/logstah --version
./bin/logstash -e 'input { stdin { } } output { stdout {} }'
直到看到"Pipeline main started"。说明启动成功。这个例子,stdin控制台输入作为数据来源,stdout作为数据结果。输入 hello可以看到2017-12-14T09:37:29.287Z chenshunyang.local hello
启动:./bin/logstash -f /home/push/logstash/conf/logstash.conf (logstash.conf为logstash的配置文件)
配置logstash:生成一个logstash.conf文件,配置示例内容如下
将数据写入es中
input {
stdin {
}
}
output {
#写出到es
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-elktest-%{+YYYY.MM}"
document_type => "elktest"
workers => 1
flush_size => 20000
idle_flush_time => 10
template_overwrite => true
}
#写出到控制台
stdout {
codec => rubydebug
}
}
filters是一个行处理机制将提供的为格式化的数据整理成你需要的数据,让我们看看下面的一个例子,叫grok filter的过滤器。elktest-filter.conf
input {
stdin {
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
}
output {
#写出到es
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-elktest-%{+YYYY.MM}"
document_type => "elktest"
workers => 1
flush_size => 20000
idle_flush_time => 10
template_overwrite => true
}
#写出到控制台
stdout {
codec => rubydebug
}
}
启动:./bin/logstash -f config/elktest-filter.conf
现在粘贴下面一行信息到你的终端(当然Logstash就会处理这个标准的输入内容):
127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"
你会看到
eg:elktest-apache.conf
input {
stdin {
}
#从文件获取数据
file {
path => "/Users/chenshunyang/tools/logs/nginx.log"
start_position => beginning
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
}
output {
#写出到es
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-elktest-%{+YYYY.MM}"
document_type => "elktest"
workers => 1
flush_size => 20000
idle_flush_time => 10
template_overwrite => true
}
#写出到控制台
stdout {
codec => rubydebug
}
}
准备好nginx.log文件
113.31.136.46 - - [14/Dec/2017:19:42:09 +0800] "POST /sendsms HTTP/1.1" 200 5 "-" "python-requests/2.18.4" "-" "0.012" "sendsms.jpushoa.com" "200" "192.168.250.220:8080" "0.012"
执行:./bin/logstash -f logstash-apache.conf 观察结果
关键配置词语解释:
Input 解释
作为 Consumer 端,插件使用的是 High-level-consumer API,请结合上述 kafka 基本概念进行设置:groupid消费者分组,可以通过组 ID 去指定,不同的组之间消费是相互不受影响的,相互隔离。topicid指定消费话题,也是必填项目,指定消费某个 topic ,这个其实就是订阅某个主题,然后去消费。resetbeginninglogstash 启动后从什么位置开始读取数据,默认是结束位置,也就是说 logstash 进程会以从上次读取结束时的偏移量开始继续读取,如果之前没有消费过,那么就开始从头读取.如果你是要导入原有数据,把这个设定改成 "true", logstash 进程就从头开始读取.有点类似 cat ,但是读到最后一行不会终止,而是变成 tail -F ,继续监听相应数据。decorateevents在输出消息的时候会输出自身的信息包括:消费消息的大小, topic 来源以及 consumer 的 group 信息。rebalancemaxretries当有新的 consumer(logstash) 加入到同一 group 时,将会 reblance ,此后将会有 partitions 的消费端迁移到新的 consumer 上,如果一个 consumer 获得了某个 partition 的消费权限,那么它将会向 zookeeper 注册, Partition Owner registry 节点信息,但是有可能此时旧的 consumer 尚没有释放此节点,此值用于控制,注册节点的重试次数。consumertimeoutms指定时间内没有消息到达就抛出异常,一般不需要改。以上是相对重要参数的使用示例,更多参数可以选项可以跟据 https://github.com/joekiller/logstash-kafka/blob/master/README.md 查看 input 默认参数。
output解释: Logstash 会努力攒到 20000 条数据一次性发送出去,但是如果 10 秒钟内也没攒够 20000 条,Logstash 还是会以当前攒到的数据量发一次。
安装filebeat:从https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.2.0-darwin-x8664.tar.gz下载解压:tar zxvf filebeat-5.2.0-darwin-x8664.tar.gz版本:./filebeat --version默认端口:5043修改filebeat.yml为elktest.yml并添加内容
filebeat:
prospectors:
- input_type: log
paths:
- /Users/chenshunyang/tools/logs/nginx.log
tail_files: true
tags: ["data-api"]
output.logstash:
hosts: ["localhost:9600"]
loadbalance: true
index: nginx-log
改为logstash接收,logstash服务端口为9600启动:./filebeat -e -c elktest.yml
修改logstash的配置,改为logstash接收filebeat的数据,并输出到控制台以及es中
input {
beats {
port => 5043
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-elktest-%{+YYYY.MM}"
document_type => "elktest"
workers => 1
flush_size => 20000
idle_flush_time => 10
template_overwrite => true
}
stdout {
codec => rubydebug
}
}
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。