问题
需求是统计平台内包括用户, 订单, 网站访问量等某时间段的各种数据, 第一想法是使用更熟悉的Mysql做数据统计, 那么有什么问题呢?
之前使用过的设计方案有两种
1. 从源表查询, 聚合使用group和count
优点: 简单
缺点:
- 速度慢, 由于数据库存储的时间是时间戳, 在group的时候还需要转化为正常的字符串时间才能按天, 月, 小时聚合.
- 性能损耗高, 可能会影响到线上业务.
2. 建立聚合表, 按天, 月, 小时
优点: 查询速度快
缺点:
- 逻辑麻烦, 难得开发.
- 数据量大, 往往造成统计数据比源数据还多的尴尬局面.
选择
有没有什么方案能解决速度慢 数据量大的问题呢?
HBase? 太重, 听说是搞大数据的, 我们小公司, 没必要了.
想到在之前有用过Graylog就是使用Elasticsearch做日志存储和统计, 那么Elasticsearch能不能做我想要的数据统计呢? 肯定是可以的.
同时ES(Elasticsearch)还能做订单和用户等数据的全文搜索, 一举多得.
当然选择ES还有一个原因就是之前了解过一点ES.
什么是Elasticsearch
Elasticsearch is a highly scalable open-source full-text search and analytics engine. It allows you to store, search, and analyze big volumes of data quickly and in near real time. It is generally used as the underlying engine/technology that powers applications that have complex search features and requirements.
Elasticsearch能近乎实时的全文搜索(如模糊搜索)和分析(如聚合)大量的数据.
使用方法
跟着官方文档先上手吧, 我选择的是5.5.1版本的Es, 使用Docker安装, 安装方法就不多说了, 按文档来就好.
这里简单讲一下Es中基本概念
index(索引)
可以理解为mysql的表
type(类型)
索引下的类型, 可以理解为分组. 在相同index的不同的type应该使用相同的数据结构.
不建议使用, 非要使用的话建议和索引同名, 因为在ES 7.x版本type会被废弃.
document(文档)
一个文档就是一条数据, 多个document组成了index
aggregation(聚合)
做统计就需要使用聚合.
可理解为mysql的group by语句, 但聚合更强大的是aggregations can be nested!(聚合可以被嵌套), 并且没有层数限制.
bucket(桶)
理解为mysql group by之后的每个条目, aggregation之后会返回多个bucket.
Date Histogram Aggregation(日期直方图聚合)
为了让按时间聚合更简单, ES内置了一个日期直方图的聚合类型, 可以方便的实现按小时(h),天(d),月(month),年(year)等时间段聚合数据.
为了使用Date Histogram Aggregation, 需要时间字段是long型(精确到毫秒的时间戳),或者是RFC3339格式的字符串(如2006-01-02T15:04:05Z07:00).
这里值得注意的是时区, 如果要设置时区, 那么时间字段必须为RFC3339格式, 所以在ES中推荐统一使用RFC3339作为时间字段类型.
由于我们的时区是+08:00, 而ES默认使用UTC, 所以如果按天聚合的话聚合时间是从每一天的8:00到下一天的8:00, 如果要重置到0:00则需要设置时区 +08:00.
{
"aggs": {
"by_day": {
"date_histogram": {
"field": "date",
"interval": "day",
"time_zone": "+08:00"
}
}
}
}
排序
假如有一个需求: 获取这个月的访问量最高的10个站点. 这时就需要使用到排序.
排序使用order
{
"aggregations": {
"pv": {
"sum": {
"field": "pv"
}
}
},
"terms": {
"field": "site_id",
"min_doc_count": 0,
"order": [
{
"pv": "desc"
}
],
"size": 10
}
}
可以指定size表示返回几个.
但不支持类似mysql的offset, 因为es不支持聚合后分页, 详见https://blog.csdn.net/laoyang360/article/details/79112946.
实战
可以用Restful接口操作ES, 但这样比较麻烦, 所以我在github上找了一个go语言的Client: gopkg.in/olivere/elastic.v5.
笔者也看了下官方包:github.com/elastic/go-elasticsearch, 但官方支持的功能太简单, 为了节约开发时间, 还是使用前者吧.
案例中使用到的结构体如下:
// 用户
type User struct {
Id int64 `json:"id"`
Nickname string `json:"nickname"`
Mobile string `json:"mobile"`
Email string `json:"email"`
CreatedAt string `json:"created_at"`
LastLoginAt int64 `json:"last_login_at"`
}
// 每天的站点访问记录, 用来聚合出每月的访问记录
type SiteVisitLog struct {
Id string `json:"id"` // es随机id
SiteId int64 `json:"site_id"` // 站点id
Day string `json:"day"` // 每天一条记录, 用每天的凌晨代表一天
Pv int64 `json:"pv"` // 每访问一次就记录
Uv int64 `json:"uv"` // 使用cookie去重, 每天(0-24h)内同一个cookie算一个
}
案例1: 统计某时间段内新用户的注册数量
使用Search即可, 与mysql不同, 查询结果中会带有命中(hit)的文档个数(TotalHits)而不需要count操作.
// getUserCount 返回start到end之间的文档数量.
func getUserCount(start int64, end int64) (item *pb.EsUserCountItem) {
// 范围查询
query := elastic.NewRangeQuery("created_at")
if start != 0 {
query.Gte(start)
}
if end != 0 {
query.Lte(end)
}
ctx := context.Background()
result, err := client.Search().
Index(user.IndexName()).
Type(user.IndexName()).
Query(query).
Size(0). // 只是统计数量, 不需要返回文档
Do(ctx)
if err != nil {
return
}
item = &pb.EsUserCountItem{All: result.Hits.TotalHits}
return
}
案例2: 聚合统计新用户数量
支持秒级别的聚合, 威力巨大.
// AggsUser聚合start到end的新增用户, interval为聚合颗粒度, 如year, quarter, month, week, day, hour, minute, second, 支持的单位详情看文档
func AggsUser(start, end int64, interval string) (items []*pb.AggsUserItem, err error) {
ctx := context.Background()
da := elastic.NewDateHistogramAggregation().
Interval(interval).
Field("created_at").
TimeZone("+08:00"). // 时区问题
MinDocCount(0) // 显示没有数据(文档数为0)的bucket(桶)
// 条件
query := elastic.NewRangeQuery("created_at")
if start != 0 {
query.Gte(start)
}
if end != 0 {
query.Lte(end)
}
result, err := client.Search().
Index(user.IndexName()).
Type(user.IndexName()).
Query(query).
Size(0).
Aggregation("a", da).
Do(ctx)
if err != nil {
return
}
bs := &structs.DateHistogramBuckets{}
err = bs.UnMarshal(result.Aggregations["a"])
if err != nil {
return
}
items = make([]*pb.AggsUserItem, len(bs.Buckets))
for i, v := range bs.Buckets {
items[i] = &pb.AggsUserItem{
Time: v.Key,
All: v.DocCount,
}
}
return
}
案例3: 聚合统计站点访问量(pv, uv)
体验嵌套聚合的威力吧.
// AggsSiteVisitLog 聚合统计站点访问量, 得到每月的总pv,uv
func AggsSiteVisitLog(siteId, start, end int64, interval string) (items []*pb.AggsSiteVisitLogItem, err error) {
ctx := context.Background()
pvAgg := elastic.NewSumAggregation().Field("pv") // 将pv求和
uvAgg := elastic.NewSumAggregation().Field("uv")
dAgg := elastic.NewDateHistogramAggregation().
Interval(interval).
Field("day").
TimeZone("+08:00").
MinDocCount(0).
SubAggregation("pv", pvAgg). // 在`时间聚合`中再嵌套`求和聚合`
SubAggregation("uv", uvAgg)
rQuery := elastic.NewRangeQuery("day")
if start != 0 {
rQuery.Gte(start)
}
if end != 0 {
rQuery.Lte(end)
}
query := elastic.NewBoolQuery()
query.Must(rQuery)
s := client.Search().
Index(siteVisitLog.IndexName()).
Type(siteVisitLog.IndexName())
if siteId != 0 {
query.Must(elastic.NewTermQuery("site_id", siteId))
}
result, err := s.
Query(elastic.NewConstantScoreQuery(query)).
Size(0).
Aggregation("a", dAgg).
Do(ctx)
if err != nil {
return
}
// 结构体有点复杂, 使用log先打印出结果再写结构体解析吧
buckets := &struct {
Buckets []*struct {
Key int64 `json:"key"`
DocCount int64 `json:"doc_count"`
Uv struct {
Value float64 `json:"value"`
} `json:"uv"`
Pv struct {
Value float64 `json:"value"`
} `json:"pv"`
} `json:"buckets"`
}{}
bs, _ := result.Aggregations["a"].MarshalJSON()
err = json.Unmarshal(bs, buckets)
if err != nil {
return
}
items = make([]*pb.AggsSiteVisitLogItem, len(buckets.Buckets))
for i, v := range buckets.Buckets {
items[i] = &pb.AggsSiteVisitLogItem{
Time: v.Key,
Uv: int64(v.Uv.Value),
Pv: int64(v.Pv.Value),
}
}
return
}
案例三有点复杂, 要知道的知识点还挺多
布尔过滤器
待续
constant_score
待续
