实时收集Storm日志到ELK集群
Posted 三劫散仙
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了实时收集Storm日志到ELK集群相关的知识,希望对你有一定的参考价值。
背景
我们的storm实时流计算项目已经上线几个月了,由于各种原因迟迟没有进行监控,每次出现问题都要登录好几台机器,然后使用sed,shell,awk,vi等各种命令来查询原因,效率非常低下,而且有些统计是没法做的,所以很有必要对storm本身相关的日志以及我们运行在storm上面的任务的日志做一个统一的日志收集,分析,查询,统计平台。
技术选型
对于这个选择,其实不用考虑那么多,借用一句名言 Life is short , You need elk ! 关于elk相关的安装这里不再介绍,可参考散仙的博客:http://qindongliang.iteye.com/category/330375
需求分析
序号 | 讨论 | 内容 |
---|---|---|
1 | storm需要收集的日志 | (1)本身的相关的日志 (2)提交任务的日志 |
2 | 日志过滤 | 排除gc的log和部分不相干业务的log |
3 | 索引如何切分 | 考虑量不是很大,按每月生成一份索引 |
4 | 索引模板定制 | 默认的动态mapping比较简答,所以我们采用自定义动态索引模板 |
5 | 日志的定期删除 | 使用es官网插件curator管理 |
核心配置
(1)es的模板定义 注意date类型的动态类型是开启docvalue的,便于快速聚合和排序
{
"order": 0,
"template": "jstorm*",
"settings": {
"index": {
"number_of_replicas": "0",
"number_of_shards": "3"
}
},
"mappings": {
"_default_": {
"dynamic_templates": [
{
"level": {
"mapping": {
"index": "not_analyzed",
"type": "string"
},
"match": "level",
"match_mapping_type": "string"
}
},
{
"message": {
"mapping": {
"index": "analyzed",
"type": "string"
},
"match": "message",
"match_mapping_type": "string"
}
},
{
"date_fields": {
"mapping": {
"doc_values": true,
"type": "date"
},
"match_mapping_type": "date"
}
},
{
"string_fields": {
"mapping": {
"index": "not_analyzed",
"type": "string"
},
"match": "*",
"match_mapping_type": "string"
}
}
],
"_all": {
"enabled": false
}
}
},
"aliases": {}
}
(2)logstash的conf定义
input{
file{
#初始化全量导入
start_position => "beginning"
#统一的storm的日志目录
path=> ["/data/logs/jstorm/**/*.log"]
#排除的路径
exclude =>["*gc*","*log_monitor*"]
#指定文件偏移量存储的文件
sincedb_path => "./sincedb"
#配置多行数据收集(针对异常)
codec => multiline {
#类似两个info之间的所有数据是一行数据
pattern => "^\[%{LOGLEVEL:loglevel}"
#true代表是两个loglevel之间的数据
#false代表两个异常之间的数据,跟上面的相反
negate=> true
#后一条的数据前面所有的,都属于这整条数据
what => "previous"
}
}
}
filter {
#使用gork直接获取日志级别和时间
grok {
match =>{"message"=>"%{LOGLEVEL:loglevel}\s*%{TIMESTAMP_ISO8601:time} "}
}
# 转化日志时间为收集的时间,并移除无用的字段
date{
match => ["time","yyyy-MM-dd HH:mm:ss.SSS","yyyy-MM-dd HH:mm:ss","ISO8601"]
remove_field => [ "time","@version" ]
}
# 这个地方可以对一些数据做过滤
# if [loglevel] == "DEBUG" {
# drop { }
# }
}
#输出到es的配置
output{
elasticsearch{
#设置索引名
index => "jstorm_pro%{+YYYY-MM}"
hosts=> ["192.168.8.5:9200","192.168.8.6:9200","192.168.8.7:9200"]
#关闭logstash自动管理模块
manage_template => false
#指定模板名为jstrom
template_name => "jstorm"
#设置flush的数量
flush_size => 3000
}
# 调试控制台输出
# stdout { codec => rubydebug }
}
辅助脚本
放在logstash的根目录下面
启动脚本:start_jstorm.sh
nohup bin/logstash -f config/jstorm.conf &> jstorm_logstash.log & echo $! >jstorm_logstash_pid&
关闭脚本:stop_jstorm.sh
kill -9 `cat jstorm_logstash_pid`
收集检索效果
一切完成后,启动logstash收集进程后,我们就可以实时在kibana里面分析数据了,非常nice!
然后,我们就可以非常快速的定位异常数据了。
以上是关于实时收集Storm日志到ELK集群的主要内容,如果未能解决你的问题,请参考以下文章
巧妙拆分bolt提升Storm集群吞吐量 增加并行处理速度 Storm & kafka处理实时日志实战topology经验谈
ELK(ElasticSearch, Logstash, Kibana)搭建实时日志分析平台