将MongoDB用户首次到达小时统计聚合查询转换为OpenSearch查询

问题描述 #

我有一段可正常运行的MongoDB聚合查询，功能是按小时统计用户首次到达的数量——同一用户若多次到访，仅在其首次到达的小时计入统计，后续到访不重复计数。

以下是带变量的MongoDB查询代码：

aggregate([
            {
                $match: {
                    sequr_code: { $in: sequr_code },
                    'actor.user.uuid': { $in: valid_users },
                    time: { $gt: start_date, $lte: end_date },
                    'location.uuid': location_id,
                },
            },
            {
                $sort: {
                    time: 1,
                },
            },
            {
                $group: {
                    _id: {
                        uuid: '$actor.user.uuid',
                    },
                    tran: { $first: '$$ROOT' },
                },
            },
            {
                $group: {
                    _id: {
                        hour: { $hour: { date: '$tran.time', timezone: location_tz } },
                        minute: {
                            $subtract: [
                                { $minute: { date: '$tran.time', timezone: location_tz } },
                                { $mod: [{ $minute: '$tran.time' }, 60] },
                            ],
                        },
                    },
                    count: { $addToSet: '$tran.actor.user.uuid' },
                },
            },
            {
                $project: {
                    interv_start: { $add: ['$_id.hour', 0] },
                    interv_end: { $add: ['$_id.hour', 1] },
                    total_user_arrivals: { $size: '$count' },
                    _id: 0,
                    uuid: '$_id.uuid',
                },
            },
        ])

我尝试多次将其转换为OpenSearch查询但未成功，恳请协助转换。预期结果为按小时（如0点、1点……23点）统计用户首次到达的数量，同一用户仅被计数一次。

示例数据如下：

{
  "actor": {
    "user": {
      "role": "ADMIN",
      "cost_center": null,
      "name": "Lauren Stephens",
      "access_groups": [],
      "employee_number": null,
      "department": null,
      "uuid": "06fa0334-6ad1-48db-af97-18410496685d",
      "email": "lauren.stephens@example.com"
    }
  },
  "location": {
    "timezone": "Asia/Kolkata",
    "name": "Ahmedabad",
    "uuid": "88daf957-cda2-45b2-b008-ff65b299cadc"
  },
  "time": "2024-04-30T23:30:00.000Z",
  "tenant": {
    "name": "Swiggy",
    "uuid": "917d3da4-705e-49b4-8d05-9d8d405db6e2"
  }
}

OpenSearch 转换方案 #

下面是完全匹配需求的OpenSearch聚合查询：

{
  "query": {
    "bool": {
      "filter": [
        { "terms": { "sequr_code": sequr_code } },
        { "terms": { "actor.user.uuid": valid_users } },
        { "range": { "time": { "gt": start_date, "lte": end_date } } },
        { "term": { "location.uuid": location_id } }
      ]
    }
  },
  "aggs": {
    // 按用户分组，提取每个用户的首次到达记录
    "user_first_arrival": {
      "terms": {
        "field": "actor.user.uuid",
        "size": 10000 // 根据实际用户数量调整，确保覆盖所有有效用户
      },
      "aggs": {
        "first_tran": {
          "top_hits": {
            "size": 1,
            "sort": [{"time": "asc"}] // 按时间升序，取第一条即为首次到达记录
          }
        }
      }
    },
    // 基于首次到达记录，按小时统计唯一用户数
    "hourly_stats": {
      "bucket_script": {
        "buckets_path": {
          "userBuckets": "user_first_arrival"
        },
        "script": "return []"
      },
      "aggs": {
        "flatten_first_arrivals": {
          "reverse_nested": {
            "path": "user_first_arrival.first_tran.hits.hits"
          }
        },
        "hourly_count": {
          "terms": {
            // 按指定时区提取小时
            "script": {
              "source": "ZonedDateTime.ofInstant(Instant.parse(doc['time'].value), ZoneId.of('" + location_tz + "')).getHour()"
            },
            "size": 24 // 固定24个小时桶
          },
          "aggs": {
            "unique_user_count": {
              "cardinality": {
                "field": "actor.user.uuid"
              }
            },
            // 格式化输出，匹配MongoDB的结果字段
            "formatted_result": {
              "bucket_script": {
                "buckets_path": {
                  "hour": "_key",
                  "count": "unique_user_count.value"
                },
                "script": {
                  "source": "return { interv_start: params.hour, interv_end: params.hour + 1, total_user_arrivals: params.count }"
                }
              }
            }
          }
        }
      }
    }
  },
  "size": 0 // 无需返回原始文档，仅获取聚合结果
}

逻辑对应说明 #

1. 过滤阶段：对应MongoDB的$match，用bool.filter实现多条件过滤，确保只处理符合要求的记录。
2. 提取首次到达记录：通过terms聚合按用户UUID分组，再用top_hits取每个用户时间最早的一条记录，对应MongoDB的$sort + $group取$first逻辑。
3. 按小时统计：用reverse_nested展平用户首次到达记录，再按指定时区提取小时分组，最后用cardinality统计唯一用户数，对应MongoDB的$group + $addToSet + $size逻辑。

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