Elasticsearch：如何通过至多两次请求实现区间覆盖值的求和统计？

文档结构 #

每个文档格式如下：

{'start': 0, 'stop': 3, 'val': 3}
{'start': 2, 'stop': 4, 'val': 1}
{'start': 5, 'stop': 6, 'val': 4}

每个文档对应x轴上从start到stop的区间（保证start < stop），并关联一个数值val。

目标 #

绘制折线图，展示x轴每个坐标点上，所有覆盖该点的文档的val值之和：

由于实际场景中文档量极大，性能要求很高，因此核心问题：

是否可以通过至多两次Elasticsearch请求实现该需求？具体如何操作？

已尝试方案 #

先通过一次ES请求获取min_start和max_stop作为x轴边界，然后将x轴划分为N个区间，对每个区间发起一次ES请求：过滤掉完全不在当前区间的文档，对val求和聚合。

该方案的问题是需要发起N+1次请求，折线图精度越高（N越大），耗时线性增长，性能很差。

已实现代码 #

N = 300  # x轴划分的区间数量

x = []
y = []

# 第一步：获取x轴边界
data = es.search(index='index_name',
                 body={
                    'aggs': {
                        'min_start': {'min': {'field': 'start'}},
                        'max_stop': {'max': {'field': 'stop'}}
                    }
                 })
min_x = data['aggregations']['min_start']['value']
max_x = data['aggregations']['max_stop']['value']

x_from = min_x
x_step = (max_x - min_x) / N
for _ in range(N):
    x_to = x_from + x_step
    
    # 对每个区间发起请求求和
    data = es.search(
        index='index_name', 
        body= {
            'size': 0,  # 不返回实际文档

            'query': {
                'bool': {
                    'should': [
                        # start在当前区间内
                        {'bool': {'must': [
                            {'range': {'start': {'gte': x_from}}},
                            {'range': {'start': {'lte': x_to}}}
                        ]}},

                        # stop在当前区间内
                        {'bool': {'must': [
                            {'range': {'stop': {'gte': x_from}}},
                            {'range': {'stop': {'lte': x_to}}}
                        ]}},

                        # 当前区间完全被start-stop覆盖
                        {'bool': {'must': [
                            {'range': {'start': {'lte': x_from}}},
                            {'range': {'stop': {'gte': x_to}}}
                        ]}}
                    ],
                    'minimum_should_match': 1  # 满足至少一个条件
                }
            },

            'aggs': {
                'vals_sum': {'sum': {'field': 'val'}}
            }
        }
    )
    
    # 记录结果
    x.append(x_from)    
    y.append(data['aggregations']['vals_sum']['value'])
    
    # 切换到下一个区间
    x_from = x_to


from matplotlib import pyplot as plt
plt.plot(x, y)

解决方案：至多两次ES请求实现 #

思路 #

利用事件点聚合的思路：每个区间[start, stop)对应两个事件：

• 在start位置，总和增加val
• 在stop位置，总和减少val

通过聚合所有这些事件点，就能得到所有关键的x坐标点及对应的总和变化。之后在客户端对事件点排序、插值计算，即可生成任意精度的折线图数据。

具体步骤 #

第一次请求：获取所有事件点与x轴边界

发起一次ES请求，通过terms聚合提取所有start和stop的取值，并对每个取值对应的val求和，同时获取x轴边界：

data = es.search(index='index_name', size=0, body={
    "aggs": {
        # 聚合所有start点的val总和
        "start_events": {
            "terms": {
                "field": "start",
                "size": 10000  # 根据实际数据量调整，确保覆盖所有不同的start值
            },
            "aggs": {
                "sum_val": {"sum": {"field": "val"}}
            }
        },
        # 聚合所有stop点的val总和
        "stop_events": {
            "terms": {
                "field": "stop",
                "size": 10000  # 同样根据实际数据量调整
            },
            "aggs": {
                "sum_val": {"sum": {"field": "val"}}
            }
        },
        # 获取x轴边界
        "x_bounds": {
            "aggs": {
                "min_start": {"min": {"field": "start"}},
                "max_stop": {"max": {"field": "stop"}}
            }
        }
    }
})

第二次请求（可选）

如果第一次请求的terms聚合因数据量过大未覆盖所有start/stop值，可单独发起一次请求获取x轴边界：

bounds_data = es.search(index='index_name', size=0, body={
    "aggs": {
        "min_start": {"min": {"field": "start"}},
        "max_stop": {"max": {"field": "stop"}}
    }
})
min_x = bounds_data['aggregations']['min_start']['value']
max_x = bounds_data['aggregations']['max_stop']['value']

客户端数据处理 #

1. 提取事件点：

   • 遍历start_events的桶，每个桶的key是start坐标，sum_val.value为要增加的数值，记为(x, +val)
   • 遍历stop_events的桶，每个桶的key是stop坐标，sum_val.value为要减少的数值，记为(x, -val)

2. 添加边界事件：

   • 在min_x处添加事件(min_x, 0)（若不存在）
   • 在max_x处添加事件(max_x, 0)（若不存在）

3. 排序事件点：按x坐标从小到大排序所有事件点。

4. 计算累计总和：

   • 初始化current_sum = 0，遍历排序后的事件点，记录每个x坐标对应的current_sum，再根据事件增减量更新current_sum。

5. 生成折线图数据：

   • 若需要指定精度（如300个点），在排序后的关键节点间进行线性插值，得到连续的x、y值。

客户端处理示例代码 #

# 提取事件点
events = []

# 处理start事件
for bucket in data['aggregations']['start_events']['buckets']:
    x = bucket['key']
    val = bucket['sum_val']['value']
    events.append( (x, val) )

# 处理stop事件
for bucket in data['aggregations']['stop_events']['buckets']:
    x = bucket['key']
    val = -bucket['sum_val']['value']
    events.append( (x, val) )

# 获取x轴边界
min_x = data['aggregations']['x_bounds']['min_start']['value']
max_x = data['aggregations']['x_bounds']['max_stop']['value']

# 添加边界事件
events.append( (min_x, 0) )
events.append( (max_x, 0) )

# 按x坐标排序
events.sort(key=lambda item: item[0])

# 计算累计总和，生成关键节点
nodes = []
current_sum = 0
prev_x = None
for x, delta in events:
    if prev_x is not None and prev_x != x:
        nodes.append( (prev_x, current_sum) )
    current_sum += delta
    nodes.append( (x, current_sum) )
    prev_x = x

# 去重相同x的节点
unique_nodes = []
seen_x = set()
for x, y_val in nodes:
    if x not in seen_x:
        seen_x.add(x)
        unique_nodes.append( (x, y_val) )

# 生成指定精度的折线数据（比如300个点）
N = 300
x_step = (max_x - min_x) / N
x_list = [min_x + i * x_step for i in range(N+1)]
y_list = []

# 插值计算每个x对应的y值
node_idx = 0
num_nodes = len(unique_nodes)
for x in x_list:
    # 找到当前x所在的区间
    while node_idx < num_nodes - 1 and unique_nodes[node_idx+1][0] < x:
        node_idx += 1
    if node_idx == num_nodes - 1:
        y_list.append(unique_nodes[-1][1])
        continue
    x1, y1 = unique_nodes[node_idx]
    x2, y2 = unique_nodes[node_idx+1]
    # 线性插值
    y = y1 if x2 == x1 else y1 + (y2 - y1) * (x - x1) / (x2 - x1)
    y_list.append(y)

# 绘图
from matplotlib import pyplot as plt
plt.plot(x_list, y_list)
plt.show()

内容的提问来源于stack exchange，提问作者Vladimir Fokow