Elasticsearch 8.X 聚合查询下的精度问题及其解决方案

1、线上环境问题

咕泡同学提问：我在看runtime文档的时候做个测试， agg求avg的时候不管是double还是long，数据都不准确，这种在生产环境中如何解决啊？

2、问题归类及出现场景

上述问题可以归类为：Elasticsearch聚合查询下的精度问题。

在日常的数据处理工作中，我们经常会遇到使用Elasticsearch进行大数据查询、统计、聚合等操作。Elasticsearch在实践中表现出优秀的搜索性能，但在一些复杂的聚合操作，如求平均值（avg）时，可能会出现数据精度不准的问题。

接下来我们将详细介绍这个问题的出现场景、可能的原因以及解决方案。

在Elasticsearch中，数据精度问题主要出现在聚合（aggregation）操作中。比如我们在做一些大数运算时，如求和（sum）、求平均值（avg），可能会遇到数据类型（double或long）导致的精度问题。这是因为Elasticsearch在进行聚合操作时，为了提高性能和效率，会使用一种叫做“浮点数计算”的方式来做大数运算，而这种计算方式在处理大数时往往会丢失一些精度。

3、问题最小化复现

以一个简单的例子来说明这个问题。我们在Elasticsearch中存储了一些商品数据，现在我们想要计算所有商品的平均价格。

数据和查询的DSL如下（已在 Elasticsearch 8.X 环境下验证过）：

数据：

POST /product/_bulk
{ "index" : { "_id" : "1" } }
{ "name" : "商品1", "price" : 1234.56 }
{ "index" : { "_id" : "2" } }
{ "name" : "商品2", "price" : 7890.12 }

查询DSL：

GET /product/_search
{
  "size": 0,
  "aggs": {
    "avg_price": {
      "avg": {
        "field": "price"
      }
    }
  }
}

虽然我们期望得到的平均价格是 (1234.56 + 7890.12) / 2 = 4562.34，但是由于浮点数计算的精度问题，返回的结果可能会略有偏差，如下图所示。

4、解决方案探讨与实现

如何解决上述聚合后精度问题呢？我们结合 Elasticsearch 基础知识和实战经验，给出如下三种解决方案。

方案一：借助 scaled_float 类型提升精度。
方案二：使用 scripted_metric 提升精度。
方案三：业务层面自己写代码实现。

接下来，我们逐一实战且解读上述三个方案。

4.1 借助 scaled_float 类型提升精度

4.1.1 什么是scaled_float？

scaled_float 是 Elasticsearch提供的一种特殊的数字数据类型，用于存储带小数的数字。

与 float 和 double 不同， scaled_float 实际上是一个 long 类型，只是它将实际的浮点数乘以一个给定的缩放因子进行存储。

在许多应用场景中，我们需要存储具有小数的数字，例如价格、评分等。float 和 double 是常用的数据类型，但它们有一些问题：例如，它们在存储和排序时可能会丢失精度，而且它们占用的存储空间比整数类型要多。而 scaled_float 实际上是将浮点数乘以一个 scaling factor，然后将结果存储为 long 。

例如，如果 scaling factor 是100，那么数字12.34将会被存储为1234。在查询和返回结果时，Elasticsearch将会除以 scaling factor ，返回原始的浮点数。

4.1.2 scaled_float的优势

精度更准确可控

与float和double相比，scaled_float在存储和排序时更准确，因为它实际上是存储的长整数，不存在浮点数的精度问题。

性能更好

由于scaled_float使用的是long类型，因此占用的存储空间更小，性能也更好。

灵活性更强

可以根据需要设置scaling factor，以平衡精度和性能。如果需要更高的精度，可以使用较大的scaling factor。如果项目需求关注性能和存储空间，可以使用较小的scaling factor。

4.1.3 在Elasticsearch中使用scaled_float

要在Elasticsearch中使用scaled_float，需要在映射中定义字段类型，并提供一个scaling factor。例如：

{
  "properties": {
    "price": {
      "type": "scaled_float",
      "scaling_factor": 100.0
    }
  }
}

这个映射定义了一个名为price的scaled_float字段，它的scaling factor是100。这意味着所有的价格都将乘以100，然后作为long存储。

例如，价格12.34将会被存储为1234。

总的来说，scaled_float是一个非常有用的工具，可以在需要存储浮点数的情况下提供更好的精度和性能。

4.1.4 实战一把，解决开篇类似问题

在这个例子中，我们有两个产品，它们的价格是浮点数。

如果想要使用scaled_float，首先需要设置一个映射（mapping）。假设想要以精确到分的精度存储价格，那么可以设置scaling_factor为100.0。以下是如何定义映射的步骤：

首先，创建一个新的索引并定义映射：

PUT /product
{
  "mappings": {
    "properties": {
      "name": {
        "type": "text"
      },
      "price": {
        "type": "scaled_float",
        "scaling_factor": 100.0
      }
    }
  }
}

这个命令创建了一个新的索引 product，并定义了两个字段：name（类型为text）和price（类型为scaled_float，scaling_factor为100.0）。

然后，批量 bulk 插入数据如下：

POST /product/_bulk
{ "index" : { "_id" : "1" } }
{ "name" : "商品1", "price" : 1234.56 }
{ "index" : { "_id" : "2" } }
{ "name" : "商品2", "price" : 7890.12 }

在这个过程中，price字段的值会自动乘以scaling_factor（在这里是100.0），然后存储为long类型。所以实际存储的值是123456和789012。

查询时，Elasticsearch会自动将价格除以scaling_factor，返回原始的浮点数。例如，如果执行以下查询：

GET /product/_doc/1

返回的结果将会是：

{
  "_index": "product",
  "_id": "1",
  "_version": 1,
  "_seq_no": 0,
  "_primary_term": 1,
  "found": true,
  "_source": {
    "name": "商品1",
    "price": 1234.56
  }
}

尽管price在存储时被乘以了100，但在查询时它又被除以了100，所以看到的价格仍然是1234.56。

这样，可以在保持较高精度的同时，使用更少的存储空间和更好的性能来存储和查询价格了。

最终咱们实现如下：

GET product/_search
{
  "size": 0,
  "aggs": {
    "avg_price": {
      "avg": {
        "field": "price"
      }
    }
  }
}

如下所示，结果精度达到预期值。

4.2 使用scripted_metric提高精度

面对这种情况，我们可以使用Elasticsearch的另一个强大功能 —— 脚本计算（scripted_metric）来解决。

scripted_metric允许我们自定义复杂的聚合逻辑，比如下面的DSL：

####务必要删除索引
DELETE product

POST /product/_bulk
{ "index" : { "_id" : "1" } }
{ "name" : "商品1", "price" : 1234.56 }
{ "index" : { "_id" : "2" } }
{ "name" : "商品2", "price" : 7890.12 }

GET /product/_search
{
  "size": 0,
  "aggs": {
    "avg_price": {
      "scripted_metric": {
        "init_script": "state.total = 0.0; state.count = 0",
        "map_script": "state.total += params._source.price; state.count++",
        "combine_script": "HashMap result = new HashMap(); result.put('total', state.total); result.put('count', state.count); return result",
        "reduce_script": """
  double total = 0.0; long count = 0; 
  for (state in states) { 
    total += state['total']; 
    count += state['count']; 
  }
  double average = total / count;
  DecimalFormat df = new DecimalFormat("#.00");
  return df.format(average);
        """
      }
    }
  }
}

Elasticsearch是一个分布式搜索和分析引擎，意味着数据可以在多个分片上存储和处理。为了处理分布式数据，Elasticsearch使用了一种名为map-reduce的编程模型。这个模型分为两个步骤：映射（Map）和归约（Reduce）。init_script，map_script，combine_script和reduce_script都是这个模型的组成部分，用于实现更复杂的聚合。

在如上的脚本中，我们定义了四个步骤：

init_script：初始化脚本，在每个分片上为每个聚合创建一个新的状态。
map_script：映射脚本，用于处理输入文档，并将其状态转化为一个可以合并的格式。
combine_script：组合脚本，用于在节点级别合并每个分片的状态。
reduce_script：归约脚本，用于在全局范围内合并状态。

通过这种方式，我们可以得到一个更精确的平均值。

上述脚本的具体含义解释如下：

init_script：这个脚本在每个分片上执行一次，为每个分片创建一个新的状态。

在上述脚本中，它创建了一个状态对象，其中包含了一个总和(total)和一个计数器(count)。这个状态对象被初始化为{total: 0.0, count: 0}。

map_script：这个脚本在每个文档上执行一次。

在上述脚本中，它读取每个文档的price字段，并将这个值添加到total，同时增加count的值。这样，total会包含所有文档价格的总和，count会包含处理过的文档数量。

combine_script：这个脚本在每个分片上执行一次，对每个分片的状态进行组合。

在上述脚本中，它只是将total和count放入一个HashMap中返回。如果有很多状态需要合并，可能会在这个脚本中进行一些预处理。

reduce_script：这个脚本在结果合并时执行一次，将所有分片的状态进行归约，计算出最终结果。

在上述脚本中，它遍历所有分片的状态，计算总的total和count，然后计算平均价格。DecimalFormat用于将平均价格格式化为两位小数的字符串。

简单来说，这就是一个分步计算平均值的过程：首先初始化状态，然后为每个文档更新状态，接着在每个分片上合并状态，最后在全局范围内合并状态并计算结果。

最终结果如下图所示，达到预期精度。

4.3 业务层面自己写代码实现。

在应用层面进行精度控制：将原始数据获取到应用层，然后在应用层进行精确的计算。这种方法的优点是可以得到非常精确的结果，但缺点是可能需要处理大量的数据，增加了网络传输和计算的负担。

在应用层面处理数据的精度问题通常需要两个步骤：

首先，需要从 Elasticsearch 获取原始数据；
然后，在应用层进行精确的计算。

以下是一个使用Java处理数据精度的例子：

假设系统应用是用 Java 编写的，可以使用 Java 的 BigDecimal 类进行精确的浮点数计算。以下是一个简单的例子：

BigDecimal price1 = new BigDecimal("1234.56");
BigDecimal price2 = new BigDecimal("7890.12");
BigDecimal average = price1.add(price2).divide(new BigDecimal(2), 2, RoundingMode.HALF_UP);

System.out.println(average);  // 输出：4562.34

上述示例中，我们首先创建了两个 BigDecimal 对象，代表两个价格。然后我们调用add方法将它们加起来，然后调用 divide 方法计算平均值。最后，我们使用 RoundingMode.HALF_UP 参数来控制舍入模式。

请注意，这种方法需要在应用层处理所有的数据，如果数据量很大，那么这可能会导致性能问题。为了减少数据传输和计算的负担，可能需要在Elasticsearch中使用更精确的查询来只获取需要的数据，或者使用Elasticsearch的聚合功能来减少返回的数据量。

此外，可能还需要在应用层做一些优化，比如使用并行处理、缓存等技术来提高处理性能。具体的方法会根据应用的具体情况和需求来决定。