elasticsearch进阶(3)—— pipeline窗口聚合函数 elasticsearch

一、参考

elasticsearch 学习系列目录——更新ing
Elasticsearch基于Pipeline窗口函数实现实时聚合计算
Moving function aggregation

二、索引数据

订单ID	订单价格	订单时间	用户
1	20	2021/09/01 01:00:00	u1
2	30	2021/09/01 01:01:00	u2
3	200	2021/09/01 01:01:30	u1
4	300	2021/09/01 01:02:00	u2
5	10	2021/09/01 01:02:30	u1
6	5	2021/09/01 01:03:00	u1
7	100	2021/09/01 01:03:30	u2
8	1000	2021/09/01 01:04:00	u2

PUT test-order/ { "mappings": { "properties": { "order_id": { "type": "keyword" }, "price": { "type": "long" }, "username": { "type": "keyword" }, "ts": { "type": "date" } } } }POST _bulk {"index":{"_index":"test-order"}} {"order_id":"1", "price": 20, "username": "u1", "ts": "2021-09-01T01:00:00Z"} {"index":{"_index":"test-order"}} {"order_id":"2", "price": 30, "username": "u2", "ts": "2021-09-01T01:01:00Z"} {"index":{"_index":"test-order"}} {"order_id":"3", "price": 200, "username": "u1", "ts": "2021-09-01T01:01:30Z"} {"index":{"_index":"test-order"}} {"order_id":"4", "price": 300, "username": "u2", "ts": "2021-09-01T01:02:00Z"} {"index":{"_index":"test-order"}} {"order_id":"5", "price": 10, "username": "u1", "ts": "2021-09-01T01:02:30Z"} {"index":{"_index":"test-order"}} {"order_id":"6", "price": 5, "username": "u1", "ts": "2021-09-01T01:03:00Z"} {"index":{"_index":"test-order"}} {"order_id":"7", "price": 100, "username": "u2", "ts": "2021-09-01T01:03:30Z"} {"index":{"_index":"test-order"}} {"order_id":"8", "price": 1000, "username": "u2", "ts": "2021-09-01T01:04:00Z"}

三、聚合查询 3.1 普通的时间聚合

GET test-order/_search { "size": 0, "query": { "range": { "ts": { "gte": "2021-09-01T01:00:00Z", "lte": "2021-09-01T01:10:00Z" } } }, "aggs": { "a1": { "date_histogram": { "field": "ts", "fixed_interval": "30s" }, "aggs": { "a2": { "sum": { "field": "price" } } } } } }{ "took" : 5, "timed_out" : false, "_shards" : { "total" : 1, "successful" : 1, "skipped" : 0, "failed" : 0 }, "hits" : { "total" : { "value" : 8, "relation" : "eq" }, "max_score" : null, "hits" : [ ] }, "aggregations" : { "a1" : { "buckets" : [ { "key_as_string" : "2021-09-01T01:00:00.000Z", "key" : 1630458000000, "doc_count" : 1, "a2" : { "value" : 20.0 } }, { "key_as_string" : "2021-09-01T01:00:30.000Z", "key" : 1630458030000, "doc_count" : 0, "a2" : { "value" : 0.0 } }, { "key_as_string" : "2021-09-01T01:01:00.000Z", "key" : 1630458060000, "doc_count" : 1, "a2" : { "value" : 30.0 } }, { "key_as_string" : "2021-09-01T01:01:30.000Z", "key" : 1630458090000, "doc_count" : 1, "a2" : { "value" : 200.0 } }, { "key_as_string" : "2021-09-01T01:02:00.000Z", "key" : 1630458120000, "doc_count" : 1, "a2" : { "value" : 300.0 } }, { "key_as_string" : "2021-09-01T01:02:30.000Z", "key" : 1630458150000, "doc_count" : 1, "a2" : { "value" : 10.0 } }, { "key_as_string" : "2021-09-01T01:03:00.000Z", "key" : 1630458180000, "doc_count" : 1, "a2" : { "value" : 5.0 } }, { "key_as_string" : "2021-09-01T01:03:30.000Z", "key" : 1630458210000, "doc_count" : 1, "a2" : { "value" : 100.0 } }, { "key_as_string" : "2021-09-01T01:04:00.000Z", "key" : 1630458240000, "doc_count" : 1, "a2" : { "value" : 1000.0 } } ] } } }

四、与flink对比 4.1 flink 创建表