golang实现Elasticsearch做短信查询统计

1.golang操作ES需要用到的第三方包

拉取地址: gopkg.in/olivere/elastic.v5 目前有v6 我用的ES是5.4.1所以用V5版本

官方有文档可以看一下API。

2.为啥要用ES做查询？

众所周知数据放进数据库(mysql.....)也可以进行查询，可以对数据进行筛选条件查询出想要的数据,那么为啥不用mysql，无非就是查询速度慢，ES查询是通过Lucene的倒排索引技术而关系型数据库用的是b-tree，b-tree索引是为写入优化的索引结构，可以下来看一下这两个技术对比一下。

3.实战我这里给个例子统计短信

func AggsSendMessage(p *pb.AggsSendMessageRq) (items *pb.AggsMessageItemRp, err error) {
    ctx := context.Background()
    costAgg := elastic.NewSumAggregation().Field("cost") // 将cost求和
    s := client.Search().
        Index(sendMessage.IndexName()).
        Index(sendMessage.IndexName())
    da := elastic.NewDateHistogramAggregation().
        Interval(p.Interval).
        Field("created_at").
        TimeZone("+08:00").
        MinDocCount(0).
        SubAggregation("cost", costAgg) // 聚合查询子查询条件
    query := elastic.NewRangeQuery("created_at")
    if p.StartTime != 0 {
        query.Gte(p.StartTime)
    }
    if p.EndTime != 0 {
        query.Lte(p.EndTime)
    }
     // 此id区分各个服务商的id，用于做查询筛选的条件
    if p.ServiceId != 0 {
        s.Query(elastic.NewTermQuery("service_id", p.ServiceId))
    }

    result, err := s.
        Query(query).
        Size(0).
        Aggregation("data", da).
        Do(ctx)
    if err != nil {
        err = errors.NewCodere(500, err, "聚合短信错误")
        return
    }
    d, err := json.Marshal(result)
    log.Printf("josn is: %s", d)
    // 把数据反序列化到bs结构里面
    bs := &DateBuckets{}
    err = bs.UnMarshal(result.Aggregations["data"])
    if err != nil {
        err = errors.NewCodere(500, err, "聚合短信错误")
        return
    }
    items = &pb.AggsMessageItemRp{}
    // key是代表月份的时间戳,DocCount代表当前key(月份)聚合出来的数量统计
    items.Item = make([]*pb.AggsMessageItem, len(bs.Buckets))
    for i, v := range bs.Buckets {
        items.Item[i] = &pb.AggsMessageItem{
            Time:  v.Key,
            Count: v.DocCount,
            Money: int64(v.Cost.Value),
        }
    }
    return
}

4.分析上面的代码

1.首先我函数传入的需求的筛选参数我就不说明了

这个函数elastic.NewSumAggregation().Field用于汇总从聚合文档中提取的数值。可以从文档中的特定数字字段提取这些值，也可以通过提供的脚本生成这些。

ES官方提供了原生的查询语句很多还是比较复杂的,封装了这些API很方便我们看一下原生查询的语句吧

{
    "query" : {
        "constant_score" : {
            "filter" : {
                "range" : { "timestamp" : { "from" : "now/1d+9.5h", "to" : "now/1d+16h" }}
            }
        }
    },
    "aggs" : {
        "intraday_return" : { "sum" : { "field" : "change" } }
    }
}

上面代码官方的例子聚合类型为求和，字段设置定义将被汇总的值的文档的数字字段。其实封装的这些API就是方便的组装成这种语句去请求,然后返回。

然后返回costAgg这个结构体存储了你的筛选条件

2.client.Search()

这个函数 是搜索的切入点,同时也返回了我定义的s变量的一个结构体存储他的查询条件

3.elastic.NewDateHistogramAggregation()

这个函数创建新的组织语法，返回一个DATE直方图聚合是一种类似于

直方图，除了它只能应用于日期值。

里面的Interval是聚合查询的间隔时间，通常是月,日,天的单位我传入的是month代表月的单位

SubAggregation()这个就是传入我们刚刚定义好的查询条件对某个字段取值

直方聚合图会根据这个条件进行筛选

MinDocCount()设置每个桶的最小文档计数

TimeZone()时区设置在计算桶之前转换日期的时区,我们+08：00个小时代表东八区+8小时，存入到字段的时候也是加了8个小时的 RFC3339 用这个格式

Field() 根据时间字段 生成直方图

4.elastic.NewRangeQuery（）

创建并初始化一个新的RangeSale.，就是将文档与具有一定范围内的术语的字段匹配。我们传入的是存储时间的一个字段created_at

5.query.Gte() 和 query.Lte()

query.Gte这个是大于等于这个的值就是查询条件 x >= ？ 这样的条件

query.Lte相反小于等于

上面代码我们查询条件 对一个时间的范围 可以这样写

6.组织好查询条件执行操作

result, err := s.
        Query(query).
        Size(0).
        Aggregation("data", da).
        Do(ctx)
    if err != nil {
        err = errors.NewCodere(500, err, "聚合短信错误")
        return
    }

上面代码s是我们上面定义好的的切入点搜索传入我们的query和da(就是我们的日期直方聚合图的条件)然后Do执行返回一个result就是查询后的结果可以序列化出来字符串打印看一下返回的json

7.result.Aggregations[data]

取出直方聚合图的搜索结果

Key      int64 `json:"key"`
    DocCount int64 `json:"doc_count"`
        Cost     Cost  `json:"cost"`

默认一个直方聚合图返回的由2个数据key代表的是一个时间戳，这是根据你查询的时候时间间隔进行筛选的我的是月(month)所以这个时间戳会标识某月,DocCount就是mouth的产生的数据有多少条，cost是我们传入的子条件查询，统计这个cost字段一个月产生多少金额(cost) ，把取出来的数据 映射到我们自己定义的数据结构中

items.Item = make([]*pb.AggsMessageItem, len(bs.Buckets))
    for i, v := range bs.Buckets {
        items.Item[i] = &pb.AggsMessageItem{
            Time:  v.Key,
            Count: v.DocCount,
            Money: int64(v.Cost.Value),
        }
    }

最后这段代码遍历挨个拿出来。

日常学习golang总结学习经验，欢迎给赞吧！有什么错误可评论我好纠正，

都是从错误中学习到东西的。