这不能用常规的聚合来实现，但是可以用script_metric aggregation来实现，它允许你实现你自己的逻辑。下面是我的尝试。代码有一部分是注解的，但是应该很简单。

POST test-events/_search
{
  "aggs": {
    "last_activities": {
      "scripted_metric": {
        "init_script": "state.docs = []",
        "map_script": "state.docs.add(new HashMap(params['_source']))",
        "combine_script": "return state.docs",
        "reduce_script": """
          def users = new HashMap(); 
          def eventTypes = new HashMap();

          for (state in states) { 
            for (d in state) { 
              def eventKey = d.event_type.toString();
              // 1. check if the event type has not been seen yet
              if (!eventTypes.containsKey(eventKey)) {
                // new event type bucket
                eventTypes[eventKey] = [
                  'key': d.event_type,
                  'docCount': 1,
                  'users': new HashSet(),
                  'uniqueUsers': 1
                ];
              } else {
                eventTypes[eventKey].docCount++;
              }
              
              // 2a. check if the user has not been seen yet
              if (!users.containsKey(d.user_id)) {
                users[d.user_id] = new HashMap(d);
              }
              // 2b. otherwise check if the event type is more recent
              else {
                if (users[d.user_id].event_type < d.event_type) {
                  def docCount = users[d.user_id].docCount;
                  users[d.user_id] = d;
                }
              }
            }
          }
          
          // add the unique user count
          users.values().stream().forEach(user -> {
            def eventKey = user.event_type.toString();
            eventTypes[eventKey].users.add(user.user_id);
            eventTypes[eventKey].uniqueUsers = eventTypes[eventKey].users.size();
          });

          eventTypes.keySet().stream().forEach(eventKey -> {
            eventTypes[eventKey].remove('users');
          });
          
          return eventTypes;
        """
      }
    }
  }
}

运行该命令得到的结果如您所料：

"aggregations" : {
    "last_activities" : {
      "value" : {
        "1" : {
          "key" : "1",
          "docCount" : 2,
          "uniqueUsers" : 1
        },
        "2" : {
          "key" : "2",
          "docCount" : 1,
          "uniqueUsers" : 1
        },
        "7" : {
          "key" : "7",
          "docCount" : 1,
          "uniqueUsers" : 1
        }
      }
    }
  }