Elasticsearch 组合聚集(Composite aggregation)实现交叉分析

实际分析应用中经常遇到需要交叉表的情况，它能够从不同维度组合分析业务。关系型数据库一般通过 group by 或 Pivot实现，很幸运，Elasticsearch也有类似功能。本文带你了解强大的组合分析，其中示例大多数来自官网。

1. 认识组合聚集

组合聚集(Composite aggregation) —— 从不同来源创建组合分组，属于多组聚集分析。与其他的多组聚集分析不同，它可以实现多聚集结果进行分页，其提供了对结果进行流化处理，类似于对文档实现滚动操作。

组合分组是从每个文档中抽取特定值进行组合，每个组合作为一个分组。举例：

{"keyword": ["foo", "bar"],"number": [23, 65, 76]
}

如果把文档中的 keyword和 number 作为值来源进行分组，结果为：

{ "keyword": "foo", "number": 23 }
{ "keyword": "foo", "number": 65 }
{ "keyword": "foo", "number": 76 }
{ "keyword": "bar", "number": 23 }
{ "keyword": "bar", "number": 65 }
{ "keyword": "bar", "number": 76 }

2. 值来源

sources 参数控制分组的数据来源，sources中定义的顺序很重要，因为它影响结果的返回顺序。每个来源的名称必须唯一，主要有三中类型的值来源。

2.1 关键词`Terms`

terms值来源相当于简单的 terms聚集，和聚集类似它的值可以从字段或脚本中抽取。举例：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "product": { "terms": { "field": "product" } } }]}}}
}

当然也可以使用脚本创建组合聚集的值来源：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{"product": {"terms": {"script": {"source": "doc['product'].value","lang": "painless"}}}}]}}}
}

2.2 直方图 `Histogram`

直方图可以为日期直方图和数值直方图。

2.2.1 数值直方图

直方图值来源可应用与数值类型构建固定大小的间隔，interval参数定义数据如何分组转换。举例，如果 interval设置为 5 ，则任何数值将转换至其最近的间隔分组，101 会被转为 100，因为其在 100 和 105之间的分组。举例：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "histo": { "histogram": { "field": "price", "interval": 5 } } }]}}}
}

同样也可以使用脚本实现：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{"histo": {"histogram": {"interval": 5,"script": {"source": "doc['price'].value","lang": "painless"}}}}]}}}
}

2.2.2 日期直方图

date_histogram 与上面 histogram类似，处理值来源为日期类型：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d" } } }]}}}
}

上面示例使用每天作为间隔转换字段 timestamp的值至最近的间隔。其他间隔参数还有 year, quarter, month, week, day, hour, minute, second。当然可以通过缩写进行表示，但不支持小数（我们可以使用90m代替 1.5h)。

日期格式

日期数据是使用64位数值进行表示，即从epoch时间以来的毫秒数。默认该数值返回作为分组的 key。我们可以通过 format参数制定日期格式：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{"date": {"date_histogram": {"field": "timestamp","calendar_interval": "1d","format": "yyyy-MM-dd"         }}}]}}}
}

时区

Elasticsearch 使用UTC格式存储日期时间，缺省所有的分组和凑整也按照UTC中完成。使用time_zone参数指示分组应该使用不同的时区。可以通过 ISO 8601 UTC （如：+01:00or-08:00），也可以使用时区ID，如：America/Los_Angeles。

偏移量

使用 offset 参数可以改变每个分组的初始值，通过制定整数 (+) 或负数 (-) 偏移量，如 1h 表示1小时， 1d 表示1天。举例，使用 day作为时间间隔，每个分组时间从午夜到午夜。设置 offset 为 +6h则改变每个分组跨度为早晨6点到下一个早晨6点：

PUT my-index-000001/_doc/1?refresh
{"date": "2015-10-01T05:30:00Z"
}PUT my-index-000001/_doc/2?refresh
{"date": "2015-10-01T06:30:00Z"
}GET my-index-000001/_search?size=0
{"aggs": {"my_buckets": {"composite" : {"sources" : [{"date": {"date_histogram" : {"field": "date","calendar_interval": "day","offset": "+6h","format": "iso8601"}}}]}}}
}

如果采用默认偏移量，则返回一个分组结果；上面的示例返回结果从6am开始，有两个分组结果：

{..."aggregations": {"my_buckets": {"after_key": { "date": "2015-10-01T06:00:00.000Z" },"buckets": [{"key": { "date": "2015-09-30T06:00:00.000Z" },"doc_count": 1},{"key": { "date": "2015-10-01T06:00:00.000Z" },"doc_count": 1}]}}
}

注：每个桶的开始偏移量是在time_zone调整之后计算的。

2.3 地理块

实际应用中暂时用不到，为了文档完整性列出标题，读者有兴趣可以参阅

2.4 组合不同来源

sources参数接收值来源数组。其可以混合不同的值来源创建组合分组，举例：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d" } } },{ "product": { "terms": { "field": "product" } } }]}}}
}

上面组合分组创建了两个值来源，date_histogram和terms ，每个分组有两个值组成，分别来自不同聚集值。任何类型的组合都可以，并且数组顺序在组合分组中保持不变。

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "shop": { "terms": { "field": "shop" } } },{ "product": { "terms": { "field": "product" } } },{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d" } } }]}}}
}

3.6 对应Java Api

通过API增加不同来源:

List<CompositeValuesSourceBuilder<?>> sources = new ArrayList<>();
sources.add(new TermsValuesSourceBuilder("field1").field("field1"));
sources.add(new TermsValuesSourceBuilder("field2").field("field2"));
sources.add(new TermsValuesSourceBuilder("field3").field("field3"));CompositeAggregationBuilder compositeAggregationBuilder =new CompositeAggregationBuilder("byProductAttributes", sources).size(10000);

设置子聚集：

compositeAggregationBuilder.subAggregation(AggregationBuilders.min("sub-field1").field("sub-field1")).subAggregation(AggregationBuilders.max("sub-field2").field("sub-field2"))
searchSourceBuilder.aggregation(compositeAggregationBuilder);

3. 其他选项

3.1 排序 (sort)

缺省分组时按照自然顺序排序，按照值的升序排列。当有多个值来源时，排序是按照每个值来源排序的，一个分组的第一个值与另一个分组的第一个进行比较，如果两者相等，则根据分组中的下一个值进行比较。这意味着 [foo, 100] 小于 [foobar, 0]，因为 foo 小于 foobar。我们可以通过给每个值来源定义排序方向，设置 order 为 asc 或 desc，举例：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d", "order": "desc" } } },{ "product": { "terms": { "field": "product", "order": "asc" } } }]}}}
}

上面示例日期直方图采用降序，而term来源分组按照升序排序。

3.2 缺失分组 (missing bucket)

给给定源中文档没有值缺省被忽略，我们也可以通过设置 missing_bucket参数为 true(默认为false)，使得响应中包括缺省值文档。

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "product_name": { "terms": { "field": "product", "missing_bucket": true } } }]}}}
}

在上面的示例中，源product_name将为product字段没有值的文档显式设置为null。sort参数指示null值应该排在第一(升序，asc)还是最后(降序，desc)。

3.3 大小 (size)

size 参数可以定义返回多个组合分组。每个组合分组作为一个分组，所以设置为 10则返回前十个分组。缺省值为10。

3.4 分页

如果组合分组数量太大(或未知)而无法在单个响应中返回，则可将检索拆分为多个请求。由于组合分组实际上是平的，所以请求的大小正好是响应中返回的分组数量(假设它们就是要返回分组的大小)。如果要检索所有组合分组，那么最好使用较小的大小(例如100或1000)，然后使用after参数检索下一个结果。例如:

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"size": 2,"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d" } } },{ "product": { "terms": { "field": "product" } } }]}}}
}

返回结果：

{..."aggregations": {"my_buckets": {"after_key": {"date": 1494288000000,"product": "mad max"},"buckets": [{"key": {"date": 1494201600000,"product": "rocky"},"doc_count": 1},{"key": {"date": 1494288000000,"product": "mad max"},"doc_count": 2}]}}
}

需要达到下一组分组，重新发送请求，使用 after参数，设置为上次返回结果中 after_key的值。举例：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"size": 2,"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d", "order": "desc" } } },{ "product": { "terms": { "field": "product", "order": "asc" } } }],"after": { "date": 1494288000000, "product": "mad max" } }}}
}

after_key通常是响应中返回的最后一个分组的键，但这并不能保证。始终使用返回的after_key，而不是从结果中删除它。

3.5 子聚集

和其他多分组聚集一样，组合聚集也支持子聚集。可以通过子聚集技术其他分组或分析每个子分组。举例，下面示例技术每个分组的平均值：

GET /_search
{"size": 0,"aggs": {"my_buckets": {"composite": {"sources": [{ "date": { "date_histogram": { "field": "timestamp", "calendar_interval": "1d", "order": "desc" } } },{ "product": { "terms": { "field": "product" } } }]},"aggregations": {"the_avg": {"avg": { "field": "price" }}}}}
}

返回结果：

{..."aggregations": {"my_buckets": {"after_key": {"date": 1494201600000,"product": "rocky"},"buckets": [{"key": {"date": 1494460800000,"product": "apocalypse now"},"doc_count": 1,"the_avg": {"value": 10.0}},{"key": {"date": 1494374400000,"product": "mad max"},"doc_count": 1,"the_avg": {"value": 27.0}},{"key": {"date": 1494288000000,"product": "mad max"},"doc_count": 2,"the_avg": {"value": 22.5}},{"key": {"date": 1494201600000,"product": "rocky"},"doc_count": 1,"the_avg": {"value": 10.0}}]}}
}

注意：当前组合分组不支持管道聚集 (pipeline聚集)。

4. 组合分组应用示例

通过上面学习，我了解了组合分组是Elasticsearch中非常强大分析特性及分页能力，下面通过示例进行实战。

需求：一家在线匹萨公司的业务数据分析。主要实现三个方面功能：对结果进行分页展示，对与每个地址创建日期直方图分析，每天匹萨的平均数量。

4.1 数据准备

首先定义mapping，为了简化，这里直接把地址设置为一个字段：

PUT /pizza
{"mappings" : {"properties": {"full_address": {"type": "keyword"},"order" : {"type" : "text"},"num_pizzas" : {"type" : "integer"},"timestamp" : {"type" : "date"}}}
}

插入示例数据：

POST /pizza/_bulk
{ "index": { "_id": 1 }}
{"full_address" : "355 First St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 2, "timestamp": "2018-04-10T12:25" }
{ "index": { "_id": 2 }}
{"full_address" : "961 Union St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 3, "timestamp": "2018-04-11T12:25" }
{ "index": { "_id": 3 }}
{"full_address" : "123 Baker St, San Francisco, CA", "order" : "vegan", "num_pizzas" : 1, "timestamp": "2018-04-18T12:25" }
{ "index": { "_id": 4 }}
{"full_address" : "1700 Powell St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 5, "timestamp": "2018-04-18T12:25" }
{ "index": { "_id": 5 }}
{"full_address" : "900 Union St, San Francisco, CA", "order" : "pepperoni", "num_pizzas" : 4, "timestamp": "2018-04-18T12:25" }
{ "index": { "_id": 6 }}
{"full_address" : "355 First St, San Francisco, CA", "order" : "pepperoni", "num_pizzas" : 3, "timestamp": "2018-04-10T12:25" }
{ "index": { "_id": 7 }}
{"full_address" : "961 Union St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 1, "timestamp": "2018-04-12T12:25" }
{ "index": { "_id": 8 }}
{"full_address" : "100 First St, San Francisco, CA", "order" : "pepperoni", "num_pizzas" : 3, "timestamp": "2018-04-11T12:25" }
{ "index": { "_id": 9 }}
{"full_address" : "101 First St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 5, "timestamp": "2018-04-11T12:25" }
{ "index": { "_id": 10 }}
{"full_address" : "355 First St, San Francisco, CA", "order" : "cheese", "num_pizzas" : 10, "timestamp": "2018-04-10T12:25" }
{ "index": { "_id": 11 }}
{"full_address" : "100 First St, San Francisco, CA", "order" : "pepperoni", "num_pizzas" : 4, "timestamp": "2018-04-11T14:25" }

4.2 分页查询

实际应用中数据量非常大，如果需要查看中间页数据，其他聚集不方便实现。

GET /pizza/_search
{"size": 0,"track_total_hits": false,"aggs": {"group_by_deliveries": {"composite": {"size": 3,"sources": [{"by_address": {"terms": {"field": "full_address"}}}]}}}
}

上面结果展示第一页，注意我们设置size参数为3：

Address	Deliveries
100 First St, San Francisco, CA	2
101 First St, San Francisco, CA	1
123 Baker St, San Francisco, CA	1

现在我们需要展示下一页结果。既然 123 Baker St, San Francisco, CA 在前面查询中是最后一条结果，因此现在查询中的参数 after设置为该值：

GET /pizza/_search
{"size" : 0,"track_total_hits" : false,"aggs" : {"group_by_deliveries": {"composite" : {"after" : { "by_address" : "123 Baker St, San Francisco, CA" },"size": 3,"sources" : [{ "by_address": { "terms" : { "field": "full_address" } } }]}} }
}

注，我们还可以设置 track_total_hits 参数的值为 false，它让Elasticsearch 在查询到足够的分组后尽快结束查询。

4.3 对每个地址创建日期直方图

这里我们设置多个源，分析每个地址每天的订单量：

GET /pizza/_search
{"size": 0,"track_total_hits": false,"aggs": {"group_by_deliveries": {"composite": {"size": 3,"sources": [{"by_address": {"terms": {"field": "full_address"}}},{"histogram": {"date_histogram": {"field": "timestamp","calendar_interval": "1d","format": "yyyy-MM-dd"}}}]}}}
}

返回结果：

{"took" : 6,"timed_out" : false,"_shards" : {"total" : 1,"successful" : 1,"skipped" : 0,"failed" : 0},"hits" : {"max_score" : null,"hits" : [ ]},"aggregations" : {"group_by_deliveries" : {"after_key" : {"by_address" : "123 Baker St, San Francisco, CA","histogram" : "2018-04-18"},"buckets" : [{"key" : {"by_address" : "100 First St, San Francisco, CA","histogram" : "2018-04-11"},"doc_count" : 2},{"key" : {"by_address" : "101 First St, San Francisco, CA","histogram" : "2018-04-11"},"doc_count" : 1},{"key" : {"by_address" : "123 Baker St, San Francisco, CA","histogram" : "2018-04-18"},"doc_count" : 1}]}}
}

上面示例给每个地址创建了一个直方图分组。我们还可以使用 after参数，查询下一页结果集：

GET /pizza/_search
{"size": 0,"track_total_hits": false,"aggs": {"group_by_deliveries": {"composite": {"size": 3,"after": {"by_address": "123 Baker St, San Francisco, CA","histogram": "2018-04-18"},"sources": [{"by_address": {"terms": {"field": "full_address"}}},{"histogram": {"date_histogram": {"field": "timestamp","calendar_interval": "1d","format": "yyyy-MM-dd"}}}]}}}
}

4.4 每天平均匹萨销售数量

为了分析每天每个地址的销售的匹萨数量，我们可以给组合聚集增加子聚集计算平均值：

GET /pizza/_search?size=0
{"track_total_hits" : false,"aggs" : {"group_by_deliveries": {"composite" : {"size": 3,"sources" : [{ "by_address": { "terms" : { "field": "full_address" } } },{ "histogram": { "date_histogram" : { "field": "timestamp", "calendar_interval": "1d" } } }]},"aggregations": {"avg_pizzas_per_day": {"avg": { "field": "num_pizzas" }}}} }
}

返回结果：

{..."aggregations" : {js"group_by_deliveries" : {"after_key" : {"by_address" : "123 Baker St, San Francisco, CA","histogram" : 1524009600000},"buckets" : [{"key" : {"by_address" : "100 First St, San Francisco, CA","histogram" : 1523404800000},"doc_count" : 2,"avg_pizzas_per_day" : {"value" : 3.5}},{"key" : {"by_address" : "101 First St, San Francisco, CA","histogram" : 1523404800000},"doc_count" : 1,"avg_pizzas_per_day" : {"value" : 5.0}},{"key" : {"by_address" : "123 Baker St, San Francisco, CA","histogram" : 1524009600000},"doc_count" : 1,"avg_pizzas_per_day" : {"value" : 1.0}}]}}
}

5. 总结

组合聚集是Elasticsearch中强大特性，可以实现对大量聚集结果进行分页，使得响应时间变得可确定。通过组合不同来源，实现对数据从不同维度进行组合分析。