golang es 聚合查询-爱代码爱编程

Es通过脚本的方式计算每条数据
go demo

	fsq := elastic.NewFunctionScoreQuery().BoostMode("replace")
	script := `if(doc['device_count'].size() != 0 && doc['device_count'].value != 0 )
							{
							double price = 0.000001;
								if (doc['discount_price'].size() != 0 && doc['discount_price'].value != 0)
								{
									price = doc['discount_price'].value;
								}
							return doc['device_count'].value * price
							}`

	fsq = fsq.AddScoreFunc(elastic.NewScriptFunction(elastic.NewScript(script)))
	
	boolQuery := elastic.NewBoolQuery()
	boolQuery.Filter(elastic.NewTermsQuery("city.code", "130200"))
	
	search := elastic.NewSearchService(es.EsClient).Index(esMediaResourceIndex).Size(10)

	search.Query(fsq.Query(boolQuery))
	searchResult, err := search.
		Pretty(true).
		Do(context.Background())

	if err != nil {
		panic(err)
	}

求最大值，最小，平均，求和

	boolQuery := elastic.NewBoolQuery()
	boolQuery.Filter(elastic.NewTermsQuery("city.code", "130200"))

	deviceSumAggs := elastic.NewSumAggregation().Field("device_count")
	deviceAvgAggs := elastic.NewAvgAggregation().Field("device_count")
	deviceMinAggs := elastic.NewMinAggregation().Field("device_count")
	deviceMaxAggs := elastic.NewMaxAggregation().Field("device_count")

	search := elastic.NewSearchService(es.EsClient).Index(esMediaResourceIndex).Size(10)
	search.Query(boolQuery)
	searchResult, err := search.
		Aggregation("device_sum", deviceSumAggs).
		Aggregation("device_avg", deviceAvgAggs).
		Aggregation("device_min", deviceMinAggs).
		Aggregation("device_max", deviceMaxAggs).
		Pretty(true).
		Do(context.Background())

	if err != nil {
		panic(err)
	}

	if deviceSumAgg, found := searchResult.Aggregations.Sum("device_sum"); found {
	    if deviceSumAgg.Value != nil {
	    	fmt.Println(*deviceSumAgg.Value)
		}
	}
	if deviceAvgAgg, found := searchResult.Aggregations.Sum("device_avg"); found {
		if deviceAvgAgg.Value != nil {
			fmt.Println(*deviceAvgAgg.Value)
		}
	}

es分桶统计计数
demo 统计不同媒体类型的设备数

	boolQuery := elastic.NewBoolQuery()
	boolQuery.Filter(elastic.NewTermsQuery("city.code", "130200"))

	mediaCodeTermAggs := elastic.NewTermsAggregation().Field("media_type.code")
	mediaCodeTermAggs.SubAggregation("device_sum", elastic.NewSumAggregation().Field("device_count"))

	search := elastic.NewSearchService(es.EsClient).Index(esMediaResourceIndex).Size(10)
	search.Query(boolQuery)
	searchResult, err := search.
		Aggregation("media_type_sum", mediaCodeTermAggs).
		Pretty(true).
		Do(context.Background())

	if err != nil {
		panic(err)
	}

	if mediaCodeTermAgg, found := searchResult.Aggregations.Terms("media_type_sum"); found {
	    for _, buckt := range mediaCodeTermAgg.Buckets {
			fmt.Println(buckt.Key)
	    	if agg, ok := buckt.Aggregations.Sum("device_sum"); ok{
	    		fmt.Println(*agg.Value)
			}
		}
	}

es 围栏的方式查询数据

package main

import (
	"context"
	"fmt"
	"github.com/olivere/elastic/v7"
	"wtp-media-house/internal/app/es"
	"wtp-media-house/internal/app/media_resource/domain/params"
	"wtp-media-house/internal/pkg/logger"
)

var (
	esIndex = "unidata_heat_with_tag_202201"
	esMediaResourceIndex = "wtp_media_resource"
)




func main() {

	genoInfos := []params.GeoFence {
		{
			Name:         "",
			GeofenceType: 1,
			Lon:          116.424268,
			Lat:          39.978495,
			Radius:       200,
			Polygon:      nil,
		},
	}

	boolQuery := elastic.NewBoolQuery()
	subBoolQuery := elastic.NewBoolQuery()
	for _, geoInfo := range genoInfos {
		query := GeofenceUniDataQuery(geoInfo)
		if query != nil {
			subBoolQuery.Should(query)
		}
	}
	boolQuery.Filter(subBoolQuery)

	search := es.EsClient.Search().Index(esMediaResourceIndex)
	search.Size(10)
	search.Query(boolQuery)

	searchResult, err := search.
		Pretty(true).
		Do(context.Background())

	if err != nil {
		panic(err)
	}

	logger.WayzLog.Info(fmt.Sprintf("【Ta特征查询】 Query took %d milliseconds ", searchResult.TookInMillis))

}

func GeofenceUniDataQuery(in interface{}) elastic.Query {
	fence, ok := in.(params.GeoFence)
	if !ok {
		return nil
	}
	fenceType := fence.GeofenceType

	//圆形
	if fenceType == 1 && fence.Lon != 0 && fence.Lat != 0 {
		return elastic.NewGeoDistanceQuery("location").
			Lat(fence.Lat).
			Lon(fence.Lon).
			Distance(fmt.Sprintf("%fm", fence.Radius))
	}

	//多边形
	if fenceType == 2 && len(fence.Polygon) != 0 {
		geoQuery := elastic.NewGeoPolygonQuery("location")
		for _, polygon := range fence.Polygon {
			if len(polygon) == 2 {
				geoQuery.AddPoint(polygon[1], polygon[0])
			}
		}
		return geoQuery
	}

	//矩形（视窗）
	if fenceType == 3 && len(fence.Polygon) == 2 {
		point1 := fence.Polygon[0]
		point2 := fence.Polygon[1]
		if len(point1) == 2 && len(point2) == 2 {
			geoQuery := elastic.NewGeoBoundingBoxQuery("location").
				TopLeft(max(point1[1], point2[1]), min(point1[0], point2[0])).
				BottomRight(min(point1[1], point2[1]), max(point1[0], point2[0]))
			return geoQuery
		}
	}
	return nil
}

func max(a, b float64) float64 {
	if a < b {
		return b
	}
	return a
}

func min(a, b float64) float64 {
	if a > b {
		return b
	}
	return a
}

es 根据输入查询的参数分桶计算
demo 查询2个城市下的信息和uv

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/olivere/elastic/v7"
	"wtp-media-house/internal/app/es"
	"wtp-media-house/internal/pkg/config"
)

func main() {
	cityCodes := []string{
		"510100", "130200",
	}

	fetch := elastic.NewFetchSourceContext(true).Include("city") // 只取部分数据

	aggs := elastic.NewFiltersAggregation()

	for _, city := range cityCodes{
		boolQuery := elastic.NewBoolQuery()
		boolQuery.Filter(elastic.NewTermsQuery("city.code", city))
		aggs.FilterWithName(city, boolQuery)
	}
	// 每个桶取2条数
	aggs.SubAggregation("city_info", elastic.NewTopHitsAggregation().FetchSourceContext(fetch).Size(2))
	// 统计每个桶的uv
	aggs.SubAggregation("city_uv", elastic.NewSumAggregation().Field("resident.uv"))
	
	search := elastic.NewSearchService(es.EsHeatClient).Index(config.ES_UNIDATA_HEAR_WITH_TAG)
	searchResult, err := search.
		Aggregation("city_code", aggs).
		Pretty(true).
		Do(context.Background())

	if err != nil {
		panic(err)
	}

	type CityInfo struct {
		City struct {
			Name string `json:"name"`
			Code string `json:"code"`
		} `json:"city"`
	}
	resultCityInfo := make(map[string][]string)
	resultCityUv := make(map[string]int)

	if agg, found := searchResult.Aggregations.Filters("city_code"); found {
		for key, bucket :=range agg.NamedBuckets {
			resultCityInfo[key] = []string{}
			if city, ok := bucket.Aggregations.TopHits("city_info"); ok {
				if city.Hits.TotalHits.Value > 0 {
				    for _, hit := range city.Hits.Hits {
				    	var cityName CityInfo
				    	json.Unmarshal(hit.Source, &cityName)
						resultCityInfo[key]= append(resultCityInfo[key], cityName.City.Name)
					}
				}
			}
			if cityUv, ok := bucket.Aggregations.Sum("city_uv"); ok {
				if cityUv.Value != nil {
					resultCityUv[key] = int(*cityUv.Value)
				}
			}
		}
	}
	fmt.Println(resultCityInfo)
	fmt.Println(resultCityUv)
}