熬夜不睡觉整理ELK技术文档，从此摆脱靠百度的工作（附源码）
之前的时候，我在自己的博客平台写过一些大概的ELK的内容，但是后面有粉丝在跟我聊天的过程中我发现有一个问题，就是很多人对于ELK的原理有误解，所以导致在使用和实现效果上都会有一些误差，尤其是一些在传统行业的朋友，在这方面更是有如此，所以我花时间进行详细的整理，但是处于篇幅的原因，我分两篇文章写这三个技术（附实践代码）希望对大家有所帮助
一、 ELK工作站简介
随着互联网的发展，数据量的指数性增长，也标志着数据时代的到来，这个时候，不仅仅是业务能力强就可以，对于数据的处理速度和使用也成为衡量一个企业的标准之一，这个时候，各种大数据技术就由此产生，而这期间，也出现了相当多的技术框架，甚至这些技术框架除大数据之外也应用到传统的项目开发中，就像今天说的ELK就是其中之一。
熬夜不睡觉整理ELK技术文档，从此摆脱靠百度的工作（附源码）
可能刚接触ELK的朋友会以为这是一门技术，但是其实不是，他是** Elasticsearch、Logstash、Kibana 三个开源软件的组合。在实时数据检索和分析场合被广泛使用，但是，后来被Elastic公司所统一，因此有了ELK这个称呼，就像NBA中OK、GDP**组合一样
而且ELK的兴起除了时代的要求之外，也有他自己本身的特性存在，下面是官方提供的他的几大特性：
处理方式灵活。
Elasticsearch 是实时全文索引，不需要像 storm 那样预先编程才能使用；
配置简易上手。
Elasticsearch 全部采用 JSON 接口，Logstash 是 Ruby DSL 设计，都是目前业界最通用的配置语法设计；
检索性能高效。
虽然每次查询都是实时计算，但是优秀的设计和实现基本可以达到全天数据查询的秒级响应；
集群线性扩展。
不管是 Elasticsearch 集群还是 Logstash 集群都是可以线性扩展的；
前端操作炫丽。
Kibana 界面上，只需要点击鼠标，就可以完成搜索、聚合功能，生成炫丽的仪表板。
好了，开始看重点，当然是从最受欢迎的ES开始介绍了
二、 ElasticSearch索引服务安装和使用
1． 简介
ElasticSearch是一个基于Lucene的搜索服务器。它提供了一个分布式多用户能力的全文搜索引擎，基于RESTful web接口。Elasticsearch是用Java开发的，并作为Apache许可条款下的开放源码发布，是当前流行的企业级搜索引擎。设计用于云计算中，能够达到实时搜索，稳定，可靠，快速，安装使用方便。
我们建立一个网站或应用程序，并要添加搜索功能，令我们受打击的是：搜索工作是很难的。我们希望我们的搜索解决方案要快，我们希望有一个零配置和一个完全免费的搜索模式，我们希望能够简单地使用JSON通过HTTP的索引数据，我们希望我们的搜索服务器始终可用，我们希望能够一台开始并扩展到数百，我们要实时搜索，我们要简单的多租户，我们希望建立一个云的解决方案。Elasticsearch旨在解决所有这些问题和更多的问题。
2． ES概念
cluster
代表一个集群，集群中有多个节点，其中有一个为主节点，这个主节点是可以通过选举产生的，主从节点是对于集群内部来说的。es的一个概念就是去中心化，字面上理解就是无中心节点，这是对于集群外部来说的，因为从外部来看es集群，在逻辑上是个整体，你与任何一个节点的通信和与整个es集群通信是等价的。
shards
代表索引分片，es可以把一个完整的索引分成多个分片，这样的好处是可以把一个大的索引拆分成多个，分布到不同的节点上。构成分布式搜索。分片的数量只能在索引创建前指定，并且索引创建后不能更改。
replicas
代表索引副本，es可以设置多个索引的副本，副本的作用一是提高系统的容错性，当某个节点某个分片损坏或丢失时可以从副本中恢复。二是提高es的查询效率，es会自动对搜索请求进行负载均衡。
recovery
代表数据恢复或叫数据重新分布，es在有节点加入或退出时会根据机器的负载对索引分片进行重新分配，挂掉的节点重新启动时也会进行数据恢复。
river
代表es的一个数据源，也是其它存储方式（如：数据库）同步数据到es的一个方法。它是以插件方式存在的一个es服务，通过读取river中的数据并把它索引到es中，官方的river有couchDB的，RabbitMQ的，Twitter的，Wikipedia的。
gateway
代表es索引快照的存储方式，es默认是先把索引存放到内存中，当内存满了时再持久化到本地硬盘。gateway对索引快照进行存储，当这个es集群关闭再重新启动时就会从gateway中读取索引备份数据。es支持多种类型的gateway，有本地文件系统（默认），分布式文件系统，Hadoop的HDFS和amazon的s3云存储服务。
discovery.zen
代表es的自动发现节点机制，es是一个基于p2p的系统，它先通过广播寻找存在的节点，再通过多播协议来进行节点之间的通信，同时也支持点对点的交互。
Transport
代表es内部节点或集群与客户端的交互方式，默认内部是使用tcp协议进行交互，同时它支持http协议（json格式）、thrift、servlet、memcached、zeroMQ等的传输协议（通过插件方式集成）。
3． 安装
1、 创建用户
es启动时需要使用非root用户，所以创建一个用户
2、 安装jdk（jdk要求1.8.20或1.7.55以上）
3、 上传es安装包
tar -zxvf elasticsearch-2.3.1.tar.gz -C /bigdata/
4、 修改配置
vi /bigdata/elasticsearch-2.3.1/config/elasticsearch.yml
#集群名称，通过组播的方式通信，通过名称判断属于哪个集群
cluster.name: bigdata
#节点名称，要唯一
node.name: es-1
#数据存放位置
path.data: /data/es/data
#日志存放位置
path.logs: /data/es/logs
#es绑定的ip地址
network.host: 172.16.0.14
#初始化时可进行选举的节点
discovery.zen.ping.unicast.hosts: ["node-4.bw.cn", "node-5.bw.cn", "node-6.bw.cn"]
5、 使用scp拷贝到其他节点
scp -r elasticsearch-2.3.1/ node-5.bw.cn:$PWD
scp -r elasticsearch-2.3.1/ node-6.bw.cn:$PWD
6、 在其他节点上修改es配置，需要修改的有node.name和network.host
7、 启动es（/bigdata/elasticsearch-2.3.1/bin/elasticsearch -h查看帮助文档）
/bigdata/elasticsearch-2.3.1/bin/elasticsearch –d
8、 用浏览器访问es所在机器的9200端口
http://172.16.0.14:9200/
{
"name" : "es-1",
"cluster_name" : "bigdata",
"version" : {
"number" : "2.3.1",
"build_hash" : "bd980929010aef404e7cb0843e61d0665269fc39",
"build_timestamp" : "2016-04-04T12:25:05Z",
"build_snapshot" : false,
"lucene_version" : "5.5.0"
},
"tagline" : "You Know, for Search"
}
取进程号并杀死对应进程
kill ps -ef | grep Elasticsearch | grep -v grep | awk '{print $2}'
4． es安装插件下载es插件
/bigdata/elasticsearch-2.3.1/bin/plugin install mobz/elasticsearch-head
#本地方式安装head插件
./plugin install file:///home/bigdata/elasticsearch-head-master.zip
#访问head管理页面
http://172.16.0.14:9200/_plugin/head
5． es的RESTful接口操作
RESTful接口URL的格式：
http://localhost:9200/<index>/<type>/[<id>]
其中index、type是必须提供的。
id是可选的，不提供es会自动生成。
index、type将信息进行分层，利于管理。
index可以理解为数据库；type理解为数据表；id相当于数据库表中记录的主键，是唯一的。
这里我用自己项目中的代码跟大家进行展示这个地方是如何实现的，大家可以作为参考在自己的pc端进行实践，搭建几台简单点的虚拟机即可
熬夜不睡觉整理ELK技术文档，从此摆脱靠百度的工作（附源码）
我的是3台机器，每台2G+20G的资源配置（你的资源够大的话也可以调整的大一些）
#向store索引中添加一些书籍
curl -XPUT 'http://172.16.0.14:9200/store/books/1' -d '{
"title": "Elasticsearch: The Definitive Guide",
"name" : {
"first" : "Zachary",
"last" : "Tong"
},
"publish_date":"2015-02-06",
"price":"49.99"
}'
#通过浏览器查询
http://172.16.0.14:9200/store/books/1
#在linux中通过curl的方式查询
curl -XGET 'http://172.16.0.14:9200/store/books/1'
#再添加一个书的信息
curl -XPUT 'http://172.16.0.14:9200/store/books/2' -d '{
"title": "Elasticsearch Blueprints",
"name" : {
"first" : "Vineeth",
"last" : "Mohan"
},
"publish_date":"2015-06-06",
"price":"35.99"
}'
# 通过ID获得文档信息
curl -XGET 'http://172.16.0.14:9200/bookstore/books/1'
#在浏览器中查看
http://172.16.0.14:9200/bookstore/books/1
# 通过_source获取指定的字段
curl -XGET 'http://172.16.0.14:9200/store/books/1?_source=title'
curl -XGET 'http://172.16.0.14:9200/store/books/1?_source=title,price'
curl -XGET 'http://172.16.0.14:9200/store/books/1?_source'
#可以通过覆盖的方式更新
curl -XPUT 'http://172.16.0.14:9200/store/books/1' -d '{
"title": "Elasticsearch: The Definitive Guide",
"name" : {
"first" : "Zachary",
"last" : "Tong"
},
"publish_date":"2016-02-06",
"price":"99.99"
}'
# 或者通过 _update API的方式单独更新你想要更新的
curl -XPOST 'http://172.16.0.14:9200/store/books/1/_update' -d '{
"doc": {
"price" : 88.88
}
}'
curl -XGET 'http://172.16.0.14:9200/store/books/1'
#删除一个文档
curl -XDELETE 'http://172.16.0.14:9200/store/books/1'
# 最简单filter查询
# SELECT * FROM books WHERE price = 35.99
# filtered 查询价格是35.99的
curl -XGET 'http://172.16.0.14:9200/store/books/_search' -d '{
"query" : {
"filtered" : {
"query" : {
"match_all" : {}
},
"filter" : {
"term" : {
"price" : 35.99
}
}
}
}
}'
#指定多个值
curl -XGET 'http://172.16.0.14:9200/store/books/_search' -d '{
"query" : {
"filtered" : {
"filter" : {
"terms" : {
"price" : [35.99, 88.88]
}
}
}
}
}'
# SELECT * FROM books WHERE publish_date = "2015-02-06"
curl -XGET 'http://172.16.0.14:9200/store/books/_search' -d '{
"query" : {
"filtered" : {
"filter" : {
"term" : {
"publish_date" : "2015-02-06"
}
}
}
}
}'
# bool过滤查询，可以做组合过滤查询
# SELECT * FROM books WHERE (price = 35.99 OR price = 99.99) AND (publish_date != "2016-02-06")
# 类似的，Elasticsearch也有 and, or, not这样的组合条件的查询方式
# 格式如下：
# {
# "bool" : {
# "must" : [],
# "should" : [],
# "must_not" : [],
# }
# }
#
# must: 条件必须满足，相当于 and
# should: 条件可以满足也可以不满足，相当于 or
# must_not: 条件不需要满足，相当于 not
curl -XGET 'http://172.16.0.14:9200/bookstore/books/_search' -d '{
"query" : {
"filtered" : {
"filter" : {
"bool" : {
"should" : [
{ "term" : {"price" : 35.99}},
{ "term" : {"price" : 99.99}}
],
"must_not" : {
"term" : {"publish_date" : "2016-02-06"}
}
}
}
}
}
}'
# 嵌套查询
# SELECT * FROM books WHERE price = 35.99 OR ( publish_date = "2016-02-06" AND price = 99.99 )
curl -XGET 'http://172.16.0.14:9200/bookstore/books/_search' -d '{
"query" : {
"filtered" : {
"filter" : {
"bool" : {
"should" : [
{ "term" : {"price" : 35.99}},
{ "bool" : {
"must" : [
{"term" : {"publish_date" : "2016-02-06"}},
{"term" : {"price" : 99.99}}
]
}}
]
}
}
}
}
}'
# range范围过滤
# SELECT * FROM books WHERE price >= 20 AND price < 100
# gt : > 大于
# lt : < 小于
# gte : >= 大于等于
# lte : <= 小于等于
curl -XGET 'http://172.16.0.14:9200/store/books/_search' -d '{
"query" : {
"filtered" : {
"filter" : {
"range" : {
"price" : {
"gt" : 20.0,
"lt" : 100
}
}
}
}
}
}'
# 另外一种 and， or， not查询
# 没有bool, 直接使用and , or , not
# 注意: 不带bool的这种查询不能利用缓存
# 查询价格既是35.99，publish_date又为"2015-02-06"的结果
curl -XGET 'http://172.16.0.14:9200/bookstore/books/_search' -d '{
"query": {
"filtered": {
"filter": {
"and": [
{
"term": {
"price":59.99
}
},
{
"term": {
"publish_date":"2015-02-06"
}
}
]
},
"query": {
"match_all": {}
}
}
}
}'
刚好讲到这里，那这里我提前引入Logstash的内容，Logstash的作用就是数据收集工具，收集来自不同地方的信息，既然ELK是一个组合，那我们来看一下LogStash是如何读取文件到es中的，后面我会用单独的一篇文章详细的讲解LogStash原理以及如何读取原理的
6． Logstash读取file写入es
input {
file {
path => "/var/nginx_logs/*.log"
codec => "json"
discover_interval => 5
start_position => "beginning"
}
}
output {
elasticsearch {
index => "flow-%{+YYYY.MM.dd}"
hosts => ["172.16.0.14:9200", "172.16.0.15:9200", "172.16.0.16:9200"]
}
}
7． Logstash+kafka+es
当数据读取进来之后，因为数据源的不同，导致产生的数据类型和内容也不同，那这个时候，能够更好的进行分类，然后对应的传输到不同的位置进行处理的话相应的会好很多，那这个时候，kafka就起到了巨大的作用，那这个地方如何进行传输分类的呢？我们看一下
input {
kafka {
type => "accesslogs"
codec => "plain"
auto_offset_reset => "smallest"
group_id => "elas1"
topic_id => "accesslogs"
zk_connect => "172.16.0.11:2181,172.16.0.12:2181,172.16.0.13:2181"
}
kafka {
type => "gamelogs"
auto_offset_reset => "smallest"
codec => "plain"
group_id => "elas2"
topic_id => "gamelogs"
zk_connect => "172.16.0.11:2181,172.16.0.12:2181,172.16.0.13:2181"
}
}
filter {
if [type] == "accesslogs" {
json {
source => "message"
remove_field => [ "message" ]
target => "access"
}
}
if [type] == "gamelogs" {
mutate {
split => { "message" => " " }
add_field => {
"event_type" => "%{message[3]}"
"current_map" => "%{message[4]}"
"current_X" => "%{message[5]}"
"current_y" => "%{message[6]}"
"user" => "%{message[7]}"
"item" => "%{message[8]}"
"item_id" => "%{message[9]}"
"current_time" => "%{message[12]}"
}
remove_field => [ "message" ]
}
}
}
output {
if [type] == "accesslogs" {
elasticsearch {
index => "accesslogs"
codec => "json"
hosts => ["172.16.0.14:9200", "172.16.0.15:9200", "172.16.0.16:9200"]
}
}
if [type] == "gamelogs" {
elasticsearch {
index => "gamelogs"
codec => plain {
charset => "UTF-16BE"
}
hosts => ["172.16.0.14:9200", "172.16.0.15:9200", "172.16.0.16:9200"]
}
}
}
