Elasticsearch中聚合sales索引嵌套字段与父文档数据并生成CSV

一、Elasticsearch 聚合查询方案 #

这是最直接的实现方式，通过过滤+嵌套聚合完成需求，后续可将结果导出为CSV。

查询DSL #

{
  "size": 0,
  "query": {
    "range": {
      "created_at": {
        "gte": "2024-01-01T00:00:00Z",
        "lte": "2024-01-31T23:59:59Z"
      }
    }
  },
  "aggs": {
    "group_by_sales_id": {
      "terms": {
        "field": "id",
        "size": 10000  // 根据实际数据量调整，确保覆盖所有唯一`sales.id`
      },
      "aggs": {
        "sales_amount": {
          "max": {  // 因每个`sales.id`对应唯一文档，用max/min/terms聚合均可
            "field": "amount"
          }
        },
        "latest_event_time": {
          "nested": {
            "path": "events"
          },
          "aggs": {
            "max_event_created_at": {
              "max": {
                "field": "events.created_at"
              }
            }
          }
        },
        "total_fees_amount": {
          "nested": {
            "path": "events"
          },
          "aggs": {
            "sum_fees": {
              "sum": {
                "field": "events.fees.amount"
              }
            }
          }
        }
      }
    }
  }
}

结果导出CSV #

1. 用curl执行查询并保存结果：

curl -X POST "http://你的ES地址:9200/sales/_search?pretty" -H "Content-Type: application/json" -d @query.json > result.json

2. 用Python脚本解析JSON生成CSV：

import json
import csv

with open('result.json', 'r') as f:
    data = json.load(f)

rows = []
for bucket in data['aggregations']['group_by_sales_id']['buckets']:
    rows.append([
        bucket['key'],
        bucket['sales_amount']['value'],
        bucket['latest_event_time']['max_event_created_at']['value_as_string'],
        bucket['total_fees_amount']['sum_fees']['value']
    ])

with open('sales_report.csv', 'w', newline='') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['sales.id', 'sales.amount', 'latest_event_created_at', 'total_fees_amount'])
    writer.writerows(rows)

二、Kibana SQL 查询方案 #

若有Kibana环境，使用SQL查询更直观，且支持直接导出CSV。

SQL 查询语句 #

SELECT
  id AS "sales.id",
  amount AS "sales.amount",
  MAX(events.created_at) AS "latest_event_created_at",
  SUM(events.fees.amount) AS "total_fees_amount"
FROM sales
WHERE created_at BETWEEN '2024-01-01' AND '2024-01-31'
GROUP BY id, amount

导出CSV步骤 #

1. 打开Kibana的Dev Tools -> Console，切换到SQL模式；
2. 执行上述SQL语句；
3. 在结果面板点击Export按钮，选择CSV格式下载即可。

三、重索引预计算方案 #

若需频繁生成此类报表，可预计算结果存储到新索引，后续直接导出。

1. 创建目标索引映射 #

PUT /sales_summary
{
  "mappings": {
    "properties": {
      "id": {"type": "keyword"},
      "amount": {"type": "float"},
      "latest_event_created_at": {"type": "date"},
      "total_fees_amount": {"type": "float"},
      "original_created_at": {"type": "date"}
    }
  }
}

2. 重索引并计算 #

POST _reindex
{
  "source": {
    "index": "sales",
    "query": {
      "range": {
        "created_at": {
          "gte": "2024-01-01T00:00:00Z",
          "lte": "2024-01-31T23:59:59Z"
        }
      }
    }
  },
  "dest": {
    "index": "sales_summary"
  },
  "script": {
    "source": """
      def latest_event = null;
      def total_fees = 0.0;
      for (event in ctx._source.events) {
        if (latest_event == null || event.created_at > latest_event) {
          latest_event = event.created_at;
        }
        if (event.fees != null && event.fees.amount != null) {
          total_fees += event.fees.amount;
        }
      }
      ctx._source.latest_event_created_at = latest_event;
      ctx._source.total_fees_amount = total_fees;
      ctx._source.original_created_at = ctx._source.created_at;
      ctx._source.remove('events');
      ctx._source.remove('status');
      ctx._source.remove('type');
    """
  }
}

3. 导出CSV #

可通过Kibana的Discover功能，选择sales_summary索引，添加所需字段后点击Export导出；或用方案一的curl+脚本方式导出。

内容的提问来源于stack exchange，提问作者Gabriel R.