如何将指定SQL转换为满足累计条件的Elasticsearch DSL查询

Got it, let's break down how to translate your SQL query into Elasticsearch DSL that handles the running total logic you need. Your original SQL calculates a cumulative sum of s_b_a for each record—specifically, summing all values for the same fk_c_c_id, same month, and where datetime is less than or equal to the current record's datetime. Your current aggregation only returns monthly totals, so we need to add pipeline aggregations to track the running total over time.

Key Concepts to Replicate the SQL Logic #

Your SQL's core requirements map to these Elasticsearch features:

• Group by fk_c_c_id: Use a terms aggregation
• Group by month (from datetime): Use a date_histogram with 1M interval
• Calculate running total for records where datetime <= current record's datetime: Use the cumulative_sum pipeline aggregation, paired with a fine-grained time histogram to capture each incremental time segment

Working Elasticsearch DSL Query #

Here's the query that replicates your SQL behavior:

POST /c_c_s_index_test/_search
{
  "size": 0,
  "aggs": {
    "group_by_customer": {
      "terms": {
        "field": "fk_c_c_id",
        "size": 10000 // Adjust based on your number of unique customers
      },
      "aggs": {
        "group_by_month": {
          "date_histogram": {
            "field": "datetime",
            "interval": "1M",
            "format": "yyyy-MM",
            "min_doc_count": 1
          },
          "aggs": {
            "sort_by_time": {
              "date_histogram": {
                "field": "datetime",
                "interval": "1d", // Use smaller interval (e.g., 1h) if your data is more granular
                "min_doc_count": 1
              },
              "aggs": {
                "daily_s_b_a_sum": {
                  "sum": {
                    "field": "s_b_a"
                  }
                },
                "running_total": {
                  "cumulative_sum": {
                    "buckets_path": "daily_s_b_a_sum"
                  }
                },
                "matching_records": {
                  "top_hits": {
                    "size": 100, // Adjust based on how many records per time segment
                    "_source": ["pk_c_c_s", "fk_c_c_id", "s_b_a", "datetime"]
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

How This Works #

Let's walk through each part:

1. group_by_customer: Groups all records by fk_c_c_id, matching the i.fk_c_c_id = t.fk_c_c_id condition in your SQL.
2. group_by_month: Splits each customer's data into monthly buckets, matching the SUBSTR(datetime, 0,7) grouping.
3. sort_by_time: Breaks each month into smaller time segments (days in this example) to track incremental sums. This ensures we can calculate the running total up to each point in time.
4. daily_s_b_a_sum: Calculates the sum of s_b_a for each daily bucket.
5. running_total: Uses the cumulative_sum pipeline aggregation to accumulate the daily sums, giving you the total of all s_b_a values from the start of the month up to that day—this replicates the i.datetime <= t.datetime logic.
6. matching_records: Uses top_hits to return the original records associated with each daily bucket, so you can link the running total back to individual entries like your SQL does.

Notes for Adjustments #

• If your datetime values are more granular (e.g., hourly), change the interval in sort_by_time to 1h instead of 1d.
• Adjust the size parameters in terms and top_hits to match your dataset's scale (too small and you'll miss data; too large and performance may suffer).
• If you need the running total at the individual document level (not per day/hour), replace the date_histogram in sort_by_time with a terms aggregation on the datetime field (but be cautious—this can generate a lot of buckets if your timestamps are highly unique).

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