很多从 MySQL 转过来的 Elasticsearch 用户总是很习惯的问一个问题：『怎么在 ES 里实现 join 操作？』过去，我们的回答一般都是：通过类似宽表的思路，将数据平铺在一个索引里。不过，最近另一家 Lucene 开发商给出了另一个方案，他们开发了一个 Elasticsearch 插件，实现了 filter 层面的 join，GitHub 项目地址见：https://github.com/sirensolutions/siren-join

不过需要提醒一下的是：filter 层面的意思，就是只相当于是 SQL 里的 exists 操作。所以目前对这个插件也不要抱有太大期望。今天我们来稍微演示一下。

安装和其他 ES 插件一样：

# bin/plugin -i solutions.siren/siren-join/1.0

注意 siren-join v1.0 只支持 ES 1.7 版本，2.0 版本支持据说正在开发中。

我们 bulk 上传这么一段数据：

{"index":{"_index":"index1","_type":"type","_id":"1"}}

{"id":1, "foreign_key":"13"}

{"index":{"_index":"index1","_type":"type","_id":"2"}}

{"id":2}

{"index":{"_index":"index1","_type":"type","_id":"3"}}

{"id":3, "foreign_key": "2"}

{"index":{"_index":"index1","_type":"type","_id":"4"}}

{"id":4, "foreign_key": "14"}

{"index":{"_index":"index1","_type":"type","_id":"5"}}

{"id":5, "foreign_key": "2"}

{"index":{"_index":"index2","_type":"type","_id":"1"}}

{"id":"1", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"2"}}

{"id":"2", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"3"}}

{"id":"3", "tag": "bbb"}

{"index":{"_index":"index2","_type":"type","_id":"4"}}

{"id":"4", "tag": "ccc"}

注意，siren-join 要求用来 join 的字段必须数据类型一致。所以，当我们要用 index2 的 id 和 index1 的foreign_key 做 join 的时候，这两个字段就要保持一致，这里为了演示，特意都改成字符串。那么我们发起一个请求如下：

# curl -s -XPOST 'http://localhost:9200/index1/_coordinate_search?pretty' -d '

{

    "query":{

        "filtered":{

            "query":{

                "match_all":{}

            },

            "filter":{

                "filterjoin":{

                    "foreign_key":{

                        "index":"index2",

                        "type":"type",

                        "path":"id",

                        "query":{

                            "terms":{

                                "tag":["aaa"]

                            }

                        }

                    }

                }

            }

        }

    },

    "aggs":{

        "avg":{

            "avg":{

                "field":"id"

            }

        }

    }

}'

意即：从 index2 中搜索 q=tag:aaa 的数据的 id，查找 index1 中对应 foreign_key 的文档的 id 数据平均值。响应结果如下：

{

    "coordinate_search" : {

        "actions" : [ {

            "relations" : {

                "from" : {

                    "indices" : [ ],

                    "types" : [ ],

                    "field" : "id"

                },

                "to" : {

                    "indices" : null,

                    "types" : null,

                    "field" : "foreign_key"

                }

            },

            "size" : 2,

            "size_in_bytes" : 20,

            "is_pruned" : false,

            "cache_hit" : true,

            "took" : 0

        } ]

    },

    "took" : 2,

    "timed_out" : false,

    "_shards" : {

        "total" : 5,

        "successful" : 5,

        "failed" : 0

    },

    "hits" : {

        "total" : 2,

        "max_score" : 1.0,

        "hits" : [ {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "5",

            "_score" : 1.0,

            "_source":{"id":5, "foreign_key": "2"}

        }, {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "3",

            "_score" : 1.0,

            "_source":{"id":3, "foreign_key": "2"}

        } ]

    },

    "aggregations" : {

        "avg" : {

            "value" : 4.0

        }

    }

}

响应告诉我们：从 index2 中搜索到 2 条参与 join 的文档，在 index1 中命中 2 条数据，最后求平均值为 4.0。

想了解更全面的 ELK Stack 知识和细节，欢迎购买我的《ELK Stack权威指南》，也欢迎加 QQ 群：315428175 哟。

ES2.0 开始提供了一个崭新的 pipeline aggregation 特性，但是 Kibana 似乎并没有立刻跟进这方面的意思，相反，Elastic 公司推出了另一个实验室产品：Timelion。
timelion 的用法在官博里已经有介绍。尤其是最近两篇如何用 timelion 实现异常告警的文章，更是从 ES 的 pipeline aggregation 细节和场景一路讲到 timelion 具体操作，我这里几乎没有再重新讲一遍 timelion 操作入门的必要了。不过，官方却一直没有列出来 timelion 支持的请求语法的文档，而是在页面上通过点击图标的方式下拉帮助。

timelion 页面设计上，更接近 Kibana3 而不是 Kibana4。比如 panel 分布是通过设置几行几列的数目来固化的；query 框是唯一的，要修改哪个 panel 的 query，鼠标点选一下 panel，query 就自动切换成这个 panel 的了。

为了方便大家在上手之前了解 timelion 能做到什么，今天特意把 timelion 的请求语法所支持的函数分为几类，罗列如下：

可视化效果类：

    .bars($width): 用柱状图展示数组

    .lines($width, $fill, $show, $steps): 用折线图展示数组

    .points(): 用散点图展示数组

    .color("#c6c6c6"): 改变颜色

    .hide(): 隐藏该数组

    .label("change from %s"): 标签

    .legend($position, $column): 图例位置

    .yaxis($yaxis_number, $min, $max, $position): 设置 Y 轴属性，.yaxis(2) 表示第二根 Y 轴

数据运算类：

    .abs(): 对整个数组元素求绝对值

    .precision($number): 浮点数精度

    .testcast($count, $alpha, $beta, $gamma): holt-winters 预测

    .cusum($base): 数组元素之和，再加上 $base

    .derivative(): 对数组求导数

    .divide($divisor): 数组元素除法

    .multiply($multiplier): 数组元素乘法

    .subtract($term): 数组元素减法

    .sum($term): 数组元素加法

    .add(): 同 .sum()

    .plus(): 同 .sum()

    .first(): 返回第一个元素

    .movingaverage($window): 用指定的窗口大小计算移动平均值

    .mvavg(): .movingaverage() 的简写

    .movingstd($window): 用指定的窗口大小计算移动标准差

    .mvstd(): .movingstd() 的简写

数据源设定类：

    .elasticsearch(): 从 ES 读取数据

    .es(q="querystring", metric="cardinality:uid", index="logstash-*", offset="-1d"): .elasticsearch() 的简写

    .graphite(metric="path.to.*.data", offset="-1d"): 从 graphite 读取数据

    .quandl(): 从 quandl.com 读取 quandl 码

    .worldbank_indicators(): 从 worldbank.org 读取国家数据

    .wbi(): .worldbank_indicators() 的简写

    .worldbank(): 从 worldbank.org 读取数据

    .wb(): .worldbanck() 的简写

以上所有函数，都在 series_functions 目录下实现，每个 js 文件实现一个 TimelionFunction 功能。

想了解更全面的 ELK Stack 知识和细节，欢迎购买我的《ELK Stack权威指南》，也欢迎加 QQ 群：315428175 哟。

ELK 收集业务日志的来源，除了应用服务器以外，还有很大一部分来自客户端。考虑到客户端网络流量的因素，一般实现上都不会要求实时上报数据，而是攒一批，等到手机连上 WIFI 网络了，再统一发送出来。所以，这类客户端日志一般都有几个特点：

1. 预先已经记录成 JSON 了；
2. 日志主体内容是一个巨大无比的数组，数据元素才是实际的单次日志记录；
3. 一次 POST 会有几 MB 到几十 MB 大小。

在处理这类数据的时候，第一关是别让数据超长直接给丢弃了（说的就是你啊，Rsyslog）；第二关就是拆分 JSON 数组，把几十 MB 数据扔 ES 字段里，显然是不利于搜索和统计需求的。今天我们就来说说怎么拆分 JSON 数组。

假设收到的是这么一段日志：

{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}

首先我们知道可以在读取的时候把 JSON 数据解析成 LogStash::Event 对象：

input {

    tcp {

        codec => json

    }

}

但是怎么把解析出来的 logs 字段拆分成多个 event 呢？这里我们可以用一个已有插件：logstash-filter-split。

filter {

    split {

        field => "logs"

    }

    date {

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

        remove_fields => ["logs", "timestamp"]

    }

}

这样，就可以得到两个 event 了：

{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","@timestamp":"2015-12-10T09:55:00Z","reason":"****"}

{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10T09:56:12Z","tracert":"****"}

看起来可能跟这个插件的文档描述不太一样。文档上写的是通过 terminator 字符，切割 field 字符串成多个 event。但实际上，field 设置是会自动判断的，如果 field 内容是字符串，就切割字符串成为数组再循环；如果内容已经是数组了，直接循环：

    original_value = event[@field]



    if original_value.is_a?(Array)

        splits = original_value

    elsif original_value.is_a?(String)

        splits = original_value.split(@terminator, -1)

    else

        raise LogStash::ConfigurationError, "Only String and Array types are splittable. field:#{@field} is of type = #{original_value.class}"

    end



    return if splits.length == 1



    splits.each do |value|

        next if value.empty?



        event_split = event.clone

        @logger.debug("Split event", :value => value, :field => @field)

        event_split[(@target || @field)] = value

        filter_matched(event_split)



        yield event_split

    end

    event.cancel

顺带提一句：这里 yield 在 Logstash 1.5.0 之前，实现有问题，生成的新事件，不会继续执行后续 filter，直接进入到 output 阶段。也就是说，如果你用 Logstash 1.4.2 来执行上面那段配置，生成的两个事件会是这样的：

{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}

{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10 17:56:12","tracert":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}

想了解更全面的 ELK Stack 知识和细节，欢迎购买我的《ELK Stack权威指南》，也欢迎加 QQ 群：315428175 哟。

前端时间折腾了一下源码调试的问题，简单总结以下。
---------------------
调试环境是window（linux理论上通用）
用到的工具类：
1:mvn：https://maven.apache.org/
elasticsearch的源码是用mvn工具管理的，根据pom.xml来下载一些依赖包非常方便。
（当然也可以用gradle，由于不太熟悉，就没研究）
安装mvn，注意配置后环境变量即可。官方文档写的很明白。
最好自己修改一下mvn的setting.xml文件中的本地repo
<!-- localRepository
   | The path to the local repository maven will use to store artifacts.
   |
   | Default: ${user.home}/.m2/repository
  <localRepository>/path/to/local/repo</localRepository>
-->
我设置成了：
<localRepository>E:/m2/repository</localRepository>
mvn -v 测试以下
2:eclipse：编辑器，应用应该还比较广泛的。我用的最新版的mars。
（intellij idea据说这是一个很牛逼的编辑器，也是因为暂时不熟悉，还没研究）
----------------------
步骤：
1: 去github上选择一个tag版本，我用的是2.1.0.
https://github.com/elastic/ela ... 2.1.0 
直接DownloadZip文件即可
（也可以用git clone下来）
解压缩。
假设目录为E:/elasticsearch-2.1.0
2: 编译源代码
cmd 打开命令行
进入源文件目录 E:/elasticsearch-2.1.0
执行 mvn package命令
这个时间段耗时比较长，当然也得根据网速情况。
会出现失败，大多是因为拉取不到依赖包。可以根据提示信息，手动去下载失败的jar，然后拷贝到本地repo对应的文件夹下边即可。
等出现build success信息的时候代表成功了。
可以到core/target目录下看到elasticsearch-2.1.0-SNAPSHOT.jar。
3:转为eclipse工程
可能习惯了eclipse工程，所以这里就直接用mvn转成了eclipse的工程，生成.classpath和.project文件。
进入core目录执行以下指令
mvn eclipse:eclipse
这一步也会消耗一些时间，通常的错误也是jar包下载不成功，根据终端打印的错误信息，把对应jar包直接下载下来，放到本地的repo对应目录下边即可，然后重新运行命令。直到成功。
之后，就会发现出现了.classpath和.project文件了。
然后打开eclipse 直接带入core中的工程即可。
4: 设置运行参数
打开刚刚导入成功的工程：
Run As----Run Configution---Args
设置ProgramArgument 为 start
设置VMArgument为 -Des.path.home=E:\elasticsearch-2.1.0\core\
完毕
-------
现在就就可以运行+调试了。

ELK Stack 在入门学习过程中，必然会碰到自己修改定制索引映射(mapping)乃至模板(template)的问题。
这时候，不少比较认真看 Logstash 文档的新用户会通过下面这段配置来制定自己的模板策略：

output {

    elasticsearch {

        host => "127.0.0.1"

        manage_template => true

        template => "/path/to/mytemplate"

        template_name => "myname"

    }

}

然而随后就发现，自己辛辛苦苦修改出来的模板，通过 curl -XGET 'http://127.0.0.1:9200/_template/myname' 看也确实上传成功了，但实际新数据索引创建出来，就是没生效！

这个原因是：Logstash 默认会上传一个名叫 logstash 的模板到 ES 里。如果你在使用上面这个配置之前，曾经运行过 Logstash（一般来说都会），那么 ES 里就已经存在这么一个模板了。你可以curl -XGET 'http://127.0.0.1:9200/_template/logstash' 验证。

这个时候，ES 里就变成有两个模板，logstash 和 myname，都匹配 logstash-* 索引名，要求设置一定的映射规则了。

ES 会按照一定的规则来尝试自动 merge 多个都匹配上了的模板规则，最终运用到索引上:https://www.elastic.co/guide/e ... lates

其中要点就是：template 是可以设置 order 参数的！而不写这个参数，默认的 order 值就是 0。order 值越大，在 merge 规则的时候优先级越高。

所以，解决这个问题的办法很简单：在你自定义的 template 里，加一行，变成这样：

{

    "template" : "logstash-*",

    "order" : 1,

    "settings" : { ... },

    "mappings" : { ... }

}

当然，其实如果只从 Logstash 配置角度出发，其实更简单的办法是：直接修改原来默认的 logstash 模板，然后模板名称也不要改，就好了：

output {

    elasticsearch {

        host => "127.0.0.1"

        manage_template => true

        template_overwrite => true

    }

}

想了解更全面的 ELK Stack 知识和细节，欢迎购买我的《ELK Stack权威指南》，也欢迎加 QQ 群：315428175 哟。