如何在Elasticsearch聚合中获取每月最后一个工作日的统计数据？

方法一：脚本计算+Top Hits聚合（实时处理） #

直接在聚合中通过Painless脚本计算每个my_date所属月份的最后一个工作日，再按wallet和月份分组，用top_hits提取目标日期的value数据：

{
  "size": 0,
  "aggs": {
    "group_by_wallet": {
      "terms": {
        "field": "wallet.keyword"
      },
      "aggs": {
        "group_by_month": {
          "date_histogram": {
            "field": "my_date",
            "calendar_interval": "month",
            "format": "yyyy-MM"
          },
          "aggs": {
            "last_workday_data": {
              "top_hits": {
                "size": 1,
                "sort": [
                  {
                    "_script": {
                      "type": "number",
                      "script": {
                        "source": """
                          // 获取当月最后一天
                          ZonedDateTime endOfMonth = doc['my_date'].value.withDayOfMonth(doc['my_date'].value.lengthOfMonth());
                          // 倒推跳过周末，找到最后一个工作日
                          while (endOfMonth.getDayOfWeek() == DayOfWeek.SATURDAY || endOfMonth.getDayOfWeek() == DayOfWeek.SUNDAY) {
                            endOfMonth = endOfMonth.minusDays(1);
                          }
                          // 计算当前日期与最后一个工作日的天数差，差值为0的就是目标文档
                          return ChronoUnit.DAYS.between(doc['my_date'].value, endOfMonth);
                        """
                      },
                      "order": "asc"
                    }
                  }
                ],
                "_source": {
                  "includes": ["my_date", "value"]
                }
              }
            }
          }
        }
      }
    }
  }
}

脚本逻辑说明 #

• 先定位my_date所在月份的最后一天
• 从最后一天倒推，跳过周六、周日，得到当月最后一个工作日
• 计算当前文档日期到该工作日的天数差，差值为0的即为要提取的目标数据
• 按差值升序排序，取第一条就是对应月份最后一个工作日的记录

方法二：预处理字段（性能优化） #

如果数据量较大，实时脚本聚合性能不足，可以在数据写入时提前计算并存储last_workday_of_month字段：

1. 创建预处理管道：

PUT _ingest/pipeline/add_last_workday
{
  "processors": [
    {
      "script": {
        "source": """
          ZonedDateTime date = ctx.my_date;
          ZonedDateTime endOfMonth = date.withDayOfMonth(date.lengthOfMonth());
          while (endOfMonth.getDayOfWeek() == DayOfWeek.SATURDAY || endOfMonth.getDayOfWeek() == DayOfWeek.SUNDAY) {
            endOfMonth = endOfMonth.minusDays(1);
          }
          ctx.last_workday_of_month = endOfMonth.toInstant().toEpochMilli();
        """
      }
    }
  ]
}

2. 写入数据时指定管道：

PUT my_index/_doc/1?pipeline=add_last_workday
{
  "my_date": "2024-03-07",
  "wallet": "wallet_1",
  "value": 100
}

3. 聚合查询时直接按预处理字段分组：

{
  "size": 0,
  "aggs": {
    "group_by_wallet": {
      "terms": {
        "field": "wallet.keyword"
      },
      "aggs": {
        "group_by_last_workday": {
          "terms": {
            "field": "last_workday_of_month",
            "format": "yyyy-MM-dd"
          },
          "aggs": {
            "total_value": {
              "sum": {
                "field": "value"
              }
            }
          }
        }
      }
    }
  }
}

补充：按月份汇总全量数据 #

如果你的需求是汇总每个wallet当月所有每日value的总和（而非仅取最后一个工作日数据），直接用普通日期直方图聚合即可：

{
  "size": 0,
  "aggs": {
    "group_by_wallet": {
      "terms": {
        "field": "wallet.keyword"
      },
      "aggs": {
        "monthly_total": {
          "date_histogram": {
            "field": "my_date",
            "calendar_interval": "month",
            "format": "yyyy-MM"
          },
          "aggs": {
            "sum_value": {
              "sum": {
                "field": "value"
              }
            }
          }
        }
      }
    }
  }
}

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