Elasticsearch嵌套子聚合实现时间区间、起终点组合与承运商ID分层统计的需求咨询

Absolutely, you can achieve this requirement entirely with nested Elasticsearch aggregations—no need to split into multiple queries! Let's walk through building the correct query step by step to match your desired output structure.

Key Approach: Layered Aggregations #

We'll stack three levels of aggregations to match your requirements:

1. Time range buckets (0-30, 30-60, 60-90 days relative to your startDate/yesterday)
2. Origin-destination city pairs grouped within each time bucket
3. Unique carrier IDs with their record counts (or record lists) grouped within each city pair

Complete Query Example #

{
  "_source": ["id", "effStartTime", "carrierId", "originData", "destinationData"],
  "size": 0, // Skip returning individual docs to optimize performance
  "query": {
    "bool": {
      "must": [
        // Filter records where startTime falls within the last 90 days up to yesterday
        {
          "range": {
            "startTime": {
              "gte": "now-90d/d",
              "lte": "now-1d/d" // Matches your "startDate = yesterday" parameter
            }
          }
        },
        // Filter by your input origin/destination cities (replace with actual values)
        {
          "term": {
            "originData": "{your_input_origin_city}"
          }
        },
        {
          "term": {
            "destinationData": "{your_input_destination_city}"
          }
        }
      ],
      "must_not": [
        { "term": { "tenderStatus": "REMOVED" } }
      ],
      "filter": [
        { "exists": { "field": "carrierId" } }
      ]
    }
  },
  "aggregations": {
    "time_buckets": {
      // 1st level: Split records into your 3 time ranges
      "range": {
        "script": {
          "source": "ChronoUnit.DAYS.between(doc['startTime'].value, params.startDate)",
          "params": {
            "startDate": "2024-05-20" // Replace with yesterday's date in yyyy-MM-dd format
          }
        },
        "ranges": [
          { "key": "0-30 days", "from": 0, "to": 30 },
          { "key": "30-60 days", "from": 30, "to": 60 },
          { "key": "60-90 days", "from": 60, "to": 90 }
        ]
      },
      "aggregations": {
        "origin_destination_pairs": {
          // 2nd level: Group by origin/destination city combinations
          "terms": {
            "script": "doc['originData'].value + '|' + doc['destinationData'].value",
            "size": 1000 // Adjust based on how many unique pairs you expect
          },
          "aggregations": {
            "carrier_summary": {
              // 3rd level: Group by unique carrier IDs and count their records
              "terms": {
                "field": "carrierId",
                "size": 1000 // Adjust based on how many unique carriers you expect
              },
              "aggregations": {
                "record_count": {
                  // Optional: Explicit count field (matches your example format)
                  "value_count": { "field": "id" }
                }
              }
            }
          }
        }
      }
    }
  }
}

Breakdown of Key Components #

• size: 0: We only care about aggregation results, so disabling individual document returns speeds up the query.
• Query Filters:
  • The range on startTime ensures we only process records from the last 90 days (covering all your time buckets) up to yesterday.
  • term queries for originData and destinationData use their keyword type for precise matching.
• Aggregation Layers:
  1. Time Buckets: Uses a script to calculate the number of days between startTime and yesterday, then groups into your three ranges.
  2. Origin-Destination Pairs: Joins the origin and destination city values into a single key for grouping.
  3. Carrier Summary: Groups each city pair by unique carrierId, with a value_count to explicitly show record counts (you can also just use the doc_count from the terms aggregation if preferred).

Optional: Return Record Lists Instead of Counts #

If you need the actual records for each carrier instead of just counts, replace the record_count aggregation with a top_hits aggregation:

"carrier_records": {
  "top_hits": {
    "size": 100, // Number of records to return per carrier
    "_source": ["id", "effStartTime", "purchaseOrderNum"]
  }
}

Notes for Compatibility #

• If your Elasticsearch version doesn't support ChronoUnit, use this alternative script to calculate days:

  "script": "(params.startDate.getTime() - doc['startTime'].value.getTime()) / (1000 * 60 * 60 * 24)"
  

• Adjust the size parameters in terms aggregations to ensure you capture all unique groups (default is 10, which is often too small).

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