output.kafka：配置要接收日志的kafka集群地址可topic名称
Kafka接收到的日志格式： {"@timestamp":"2018-08-07T09:36:00.140Z","beat":{"hostname":"db-7eb166d3","name":"10.63.144.71","version":"5.4.0"},"input_type":"log","message":"# User@Host: select[select] @ [10.63.144.16] Id: 23460596\n# Query_time: 0.155956 Lock_time: 0.000079 Rows_sent: 112 Rows_examined: 366458\nSET timestamp=1533634557;\nSELECT DISTINCT(uid) FROM common_member WHERE hideforum=-1 AND uid != 0;","offset":1753219021,"source":"/data/slow/mysql_slow.log","type":"log"} Logstash配置logstash完整的配置文件如下：
input { kafka { bootstrap_servers => "10.82.9.202:9092,10.82.9.203:9092,10.82.9.204:9092" topics => ["mysql_slowlog_v2"] } } filter { json { source => "message" } grok { # 有ID有use match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id:\s%{NUMBER:id:int}\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nuse\s(?<dbname>\w+);\nSET\s+timestamp=%{NUMBER:timestamp_mysql:int};\n(?<query>.*)" ] # 有ID无use match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id:\s%{NUMBER:id:int}\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nSET\s+timestamp=%{NUMBER:timestamp_mysql:int};\n(?<query>.*)" ] # 无ID有use match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nuse\s(?<dbname>\w+);\nSET\s+timestamp=%{NUMBER:timestamp_mysql:int};\n(?<query>.*)" ] # 无ID无use match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nSET\s+timestamp=%{NUMBER:timestamp_mysql:int};\n(?<query>.*)" ] } date { match => ["timestamp_mysql","UNIX"] target => "@timestamp" } } output { elasticsearch { hosts => ["10.82.9.208:9200","10.82.9.217:9200"] index => "mysql-slowlog-%{+YYYY.MM.dd}" } } 重要参数解释：input：配置kafka的集群地址和topic名字
filter：过滤日志文件，主要是对message信息（看前文kafka接收到的日志格式）进行拆分，拆分成一个一个易读的字段，例如User、Host、Query_time、Lock_time、timestamp等。grok段根据我们前文对mysql慢日志的分类分别写不通的正则表达式去匹配，当有多条正则表达式存在时，logstash会从上到下依次匹配，匹配到一条后边的则不再匹配。date字段定义了让SQL中的timestamp_mysql字段作为这条日志的时间字段，kibana上看到的实践排序的数据依赖的就是这个时间
output：配置ES服务器集群的地址和index，index自动按天分割
kibana查询展示打开Kibana添加mysql-slowlog-*的Index，并选择timestamp，创建Index Pattern
进入Discover页面，可以很直观的看到各个时间点慢日志的数量变化，可以根据左侧Field实现简单过滤，搜索框也方便搜索慢日志，例如我要找查询时间大于2s的慢日志，直接在搜索框输入query_time: > 2回车即可
点击每一条日志起边的很色箭头能查看具体某一条日志的详情