ElasticSearch Python客户端聚合查询size与aggs相关报错问题求解

报错原因 #

1. helpers.scan使用错误 #

helpers.scan是专门用于批量拉取大量原始文档的工具，底层依赖scroll接口遍历查询命中的原始数据，本身不支持聚合查询。同时scan方法内部会自动传递size参数，和你在查询DSL中声明的size:0冲突，就会触发search() got multiple values for keyword argument 'size'报错。即使你删掉DSL里的size配置，scan也只会返回原始文档，不会处理聚合结果，自然聚合无法正常运行。

2. search接口传参错误 #

7.x版本的Python Elasticsearch客户端对search方法的参数做了拆分：query参数仅接收查询DSL中query字段的内容，aggs、size等都属于独立的顶层参数。你把包含aggs、size的完整DSL全部传入query参数，ES解析时会在query结构中识别到未知的aggs字段，就会报unknown query [aggs]的400错误。

解决方案 #

aggs是ES Search API的合法顶层参数，只需要调整传参方式即可，有两种可行写法：

写法1：按客户端规范拆分参数 #

# 拆分原DSL的各部分
query_part = {
  "bool": {
    "filter": [
      {
        "match_all": {}
      },
      {
        "range": {
          "endTime": {
            "gte": "2021-11-01T00:00:00Z",
            "lt": "2021-11-03T00:00:00Z"
          }
        }
      }
    ]
  }
}

aggs_part = {
  "SHAID": {
    "terms": {
      "field": "identiferid",
      "order": {
        "sort": "desc"
      }
    },
    "aggs": {
      "update": {
        "date_histogram": {
          "field": "endTime",
          "calendar_interval": "1d"
        },
        "aggs": {
          "update1": {
            "sum": {
              "script": {
                "lang": "painless",
                "source":"""
                    if (doc['distanceIndex.att'].size()!=0) { 
                        return doc['distanceIndex.att'].value;
                    } 
                    else { 
                        if (doc['distanceIndex.att2'].size()!=0) { 
                        return doc['distanceIndex.att2'].value;
                    }
                    return null;
                    }
                    """
              }
            }
          },
          "update2": {
            "sum": {
              "script": {
                "lang": "painless",
                "source":"""
                    if (doc['distanceIndex.att3'].size()!=0) { 
                        return doc['distanceIndex.att3'].value;
                    } 
                    else { 
                        if (doc['distanceIndex.at4'].size()!=0) { 
                        return doc['distanceIndex.att4'].value;
                    }
                    return null;
                    }
                    """
              }
            }
          }
        }
      },
      "sort": {
        "sum": {
          "field": "time2"
        }
      }
    }
  }
}

# 调用search接口
resp = es_client.search(
    index="test_index",
    query=query_part,
    aggs=aggs_part,
    size=0,
    request_timeout=1500
)

# 聚合结果存储在resp['aggregations']中
aggs_result = resp['aggregations']

写法2：直接传入完整DSL #

如果你习惯写完整的DSL结构，可以用body参数传递整个查询，不需要拆分：

# search_query就是你原来写的完整DSL
resp = es_client.search(
    index="test_index",
    body=search_query,
    request_timeout=1500
)

内容的提问来源于stack exchange，提问作者blue-sky