基本框架：

filebeat->logstash->elasticsearch

版本：

filebeat:5.4.3

logstash:5.4.3

elasticsearch:5.4.3

安装方式：

官网下载rpm包，也可下载二进制文件解压

filebeat配置：

vim /etc/filebeat/filebeat.yml

filebeat.prospectors:

• input_type: log
  paths:
  • /usr/local/mysql/slow_query.log
    document_type: mysqlslowquerylog

output.logstash:

hosts: ["[IP]:[5045]"]

filebeat.idle_timeout: 3s

logstash配置：

vim /etc/logstash/conf.d/logstash.slowquery.conf

input {

beats {

port => "5045"

codec => multiline {

pattern => "^# User@Host:"

negate => true

what => "previous"

}

}

}

filter {

grok {

match => { "message" => "(?m)^#\s+User@Host:\s+%{USER:user}[[^\]]+]\s+@\s+(?:(?\S) )?[(?:%{IPV4:clientip})?]\s+Id:\s+%{NUMBER:row_id:int}\n#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Loc

k_time:\s+%{NUMBER:lock_time:float}\s+Rows_sent:\s+%{NUMBER:rows_sent:int}\s+Rows_examined:\s+%{NUMBER:rows_examined:int}\n\s(?:use %{DATA:database};\s\n)?SET\s+timestamp=%{NUMBER:timestamp};\n\s(?(?\w+)\b.;)\s(?:\n#\s+Time)?.*$" }

}

date {

match => [ "timestamp", "UNIX", "YYYY-MM-dd HH:mm:ss"]

remove_field => [ "timestamp" ]

}

}

output {stdout{codec => rubydebug}}

output {

elasticsearch {

hosts => [ "[IP]:9200" ]

index => "mysql.slowquery"

}

}

启动：

/etc/init.d/filebeat start

nohub /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/ &

/etc/init.d/elasticsearch start