能否在Elasticsearch中添加自定义聚合字段？含Postgres替代方案

Elasticsearch 端实现 #

1. 索引时自动计算并存储 total_transaction #

可以通过Ingest Pipeline在文档写入ES前自动计算嵌套交易的总价，将结果作为字段存入文档：

创建处理管道

PUT _ingest/pipeline/customer_total_pipeline
{
  "processors": [
    {
      "script": {
        "source": """
          double total = 0;
          for (def tx : ctx.transactions) {
            total += Double.parseDouble(tx.price);
          }
          ctx.total_transaction = total;
        """
      }
    }
  ]
}

使用管道

• 写入文档时指定管道：

PUT customer_index/_doc/1?pipeline=customer_total_pipeline
{
  "customer_id": 101,
  "name": "John Doe",
  "transactions": [
    {"product_id":11,"product_name":"T-Shirt","transaction_id":"TX101","price":"500"},
    {"product_id":11,"product_name":"T-Shirt","transaction_id":"TX101","price":"600"},
    {"product_id":12,"product_name":"Shirt","transaction_id":"TX102","price":"1000"}
  ]
}

• 或设置为索引默认管道，后续写入自动生效：

PUT customer_index/_settings
{
  "index.default_pipeline": "customer_total_pipeline"
}

2. 检索时动态计算（不存储字段） #

如果不想修改存储的文档，可在查询时用Script Fields实时计算总价，仅在返回结果中包含该字段：

GET customer_index/_search
{
  "query": {
    "match_all": {}
  },
  "script_fields": {
    "total_transaction": {
      "script": {
        "source": """
          double total = 0;
          for (def tx : ctx._source.transactions) {
            total += Double.parseDouble(tx.price);
          }
          return total;
        """
      }
    }
  }
}

若transactions是嵌套类型（nested），需确保脚本能正确访问原始字段，上述写法直接读取ctx._source可兼容嵌套结构。

PostgreSQL 端实现 #

在同步数据到ES前，直接在Postgres中计算好总价，生成包含嵌套交易数组和总价的结果集，再同步到ES即可：

假设三张表结构为：

• customers：customer_id, name
• products：product_id, product_name
• transactions：transaction_id, customer_id, product_id, price

执行查询：

SELECT
  c.customer_id,
  c.name,
  JSON_AGG(
    JSON_BUILD_OBJECT(
      'product_id', p.product_id,
      'product_name', p.product_name,
      'transaction_id', t.transaction_id,
      'price', t.price::TEXT
    )
  ) AS transactions,
  SUM(t.price) AS total_transaction
FROM customers c
JOIN transactions t ON c.customer_id = t.customer_id
JOIN products p ON t.product_id = p.product_id
GROUP BY c.customer_id, c.name;

该查询会直接返回每个客户的完整数据（含交易数组和总价），后续将结果同步到ES即可。

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