ELK之logstash单节点安装

最近在搞ELK，写个文章记录分享一下经验

去官网上下载对应版本的logstash安装包，将工具包上传至服务器。

1.解压工具包

命令：

tar -xzvf logstash.tar.gz

2.配置logstash的配置

话不多说，直接上配置文件

input {
	#从kafka获取数据，还支持从es、filebeat等获取数据
	kafka {
		codec => "json"
		group_id => "testtopic"
		#此处的topics_pattern => 应该是需要和filebeat中配置的topic一致。
		topics_pattern => "testtopic"
		#此处配置Kafka服务器集群地址
		bootstrap_servers => "localhost:9092"
		auto_offset_reset => "latest"
	}
}
filter {
	
	#匹配数据
	grok {
		match => { "message" => "(?m)(?<request_id>\[(.*)\] - )(?<test-data>\{\"(.*)\"\})" }
	}
	#日期格式化
	date {
        match => [ "date", "yyyy-MM-dd" ]
    }
	#将匹配的数据转成json
	json {
      source => "test-data"
    }
	mutate {
	  #丢弃的数据
      remove_field => [ "agent", "log", "@version", "host", "input", "ecs"]
	  #替换key，若是这个key在json中不存在，则添加一个，还有个update没有添加功能
	  replace => {"request.响应参数" => "1234567"}
	  #修改字段的key，[][]该种方式是修改对象中的字段
	  rename => ["request", "req_param"]
	  rename => ["[header][request_id]", "request_id"]
	  # 将request_id中的[字符换成空
	  gsub => [ 
	           "request_id", "\[", ""
			 ]
    }
	#从es中查询数据，在output时进行判断合并文档
	elasticsearch {
		hosts => ["localhost:9200"]
		index => "test-data-%{date}"
		#查询sql
		query => "request_id:%{request_id}"
		#需要从数据库中查询的字段（request_id），需要放的event中的命名为（db_request_id）
		fields => { "request_id" => "db_request_id"}
		#需要从数据库配置中查询的字段（_id），需要放的event中的命名为（document_id）
		docinfo_fields => {
            "_id" => "document_id"
            "_index" => "document_index"
        }
	}
}
output {
	#输出到es
	if [request_id] == [db_request_id]{
	    #更新操作
		elasticsearch {
			hosts => ["localhost:9200"]
			index => "%{document_index}"
			document_id => "%{document_id}"
			##es要执行的动作 index, delete, create, update
			action => "update"
			doc_as_upsert => true
		}
	} else {
		elasticsearch {
			hosts => ["localhost:9200"]
			index => "test-data-%{date}"
		}
	}
	
	#输出到kafka
	kafka {
        bootstrap_servers => "localhost:9092"
        topic_id => "testtopic_after"
        codec => plain {
			format => "%{message}"
		}
    }
	
	stdout { codec => "rubydebug" }
}

更新操作说明：

• 必须指定document_id ，如果不指定，无法启动pipeline
• 如果 document_id 在es中不存在，会创建一条新的文档，使用document_id 作为该文档的_id。
• 如果 document_id 在es中存在，直接更新该文档。
• 有document_id 的情况下：

doc_as_upsert为true，使用event的值做为文档的值，相同的key会被覆盖，已经存在的key不会丢失
scripted_upsert为true，使用script作为文档的值
其余情况，使用upsert作为文档的值

3.启动logstash

nohup bin/logstash -f logstash.conf &

配置信息参考网址：
语法配置
grok测试网址