multi-stage aggregation pipeline matching data based on fields retrieved through $lookup

user_5080858 0
Viewed 376

I'm trying to build a complex, nested aggregation pipeline in MongoDB (4.4.9 Community Edition, using the pymongo driver for Python 3.10).

There are relevant data points in different collections which I want to aggregate into one, NEW (ideally) view (or, if that doesn't work) collection.

The collections, and the relevant fields therein follow a hierarchy. There is members, which contains the top-level key on which other data is to be merged, membershipNumber.

> members.find_one()
{'_id': ObjectId('61153299af6122XXXXXXXXXXXXX'), 'membershipNumber': 'N03XXXXXX'}

Then, there's a different collection, which contains membershipNumber, but also a different, linked field, an_user_id. an_user_id is used in other collections to denote records/fields in arrays that pertain to that particular user.

I 'join' members and an_users like so:

result = members.aggregate([
    {
        '$lookup': {
            'from': 'an_users',                   
            'localField': 'membershipNumber',     
            'foreignField': 'memref',             
            'as': 'an_users'                      
        }
    },
    {   '$unwind' : '$an_users' },     
    {   
        '$project' : {
            '_id' : 1,
            'membershipNumber' : 1,
            'an_user_id' : '$an_users.user_id'
        } 
    }
]);

So far so good, this returns the desired, aggregated record:

{'_id': ObjectId('61153253aBBBBBBBBBBBB'),
  'membershipNumber': 'N0XXXXXXXX',
  'an_user_id': '48XXXXXX'}

Now, I have a third collection, which contains the an_user_id as a string in arrays, denoting wherever that user clicked a given email, whereby a record is an email (and the an_user_ids in the clicks array are users that clicked a link in that email.

{'_id': ObjectId('blah'),
 'email_id': '407XXX',
 'actions_count': 17,
 'administrative_title': 'test',
 'bounce': ['3440XXXX'],
 'click': ['38294CCC',
  '418FFFF',
  '48XXXXXX',
  '38eGGGG'}

I want to count the number occurences of a given an_user_id (which I've attained from aggregating) in arrays (e.g. clicks, bounces, opens) in the emails collection, and include it in the .aggregate call, to retrieve something like this:

{'_id': ObjectId('61153253aBBBBBBBBBBBB'),
  'membershipNumber': 'N0XXXXXXXX',
  'an_user_id': '48XXXXXX',
  'n_email_clicks' : 412,
  'n_email_bounces' : 12
}

Further, I might want to also attach counts of an_user_id in other collections in my DB.

Consider, e.g., this collection called events:

{
    "_id": "617ffa96ee11844e143a63dd",
    "id": "12345",
    "administrative_title": "my_event",
    "created_at": {
        "$date": "2020-01-15T16:28:50.000Z"
    },
    "event_creator_id": "123456",
    "event_title": "my_event",
    "group_id": "123456",
    "permalink": "event_id",
    "rsvp_count": 54,
    "rsvps": [{
        "rsvp_id": "56789",
        "display_name": "John Doe",
        "rsvp_user_id": "48XXXXXX",
        "rsvp_created_at": {
            "$date": "2020-01-28T15:38:50.000Z"
        },
        "rsvp_updated_at": {
            "$date": "2020-01-28T15:38:50.000Z"
        },
        "first_name": "John",
        "last_name": "Doe",
    }, {
        "rsvp_id": "543895",
        "display_name": "James Appleslice",
        "rsvp_user_id": "N03XXXXXX",
        "rsvp_created_at": {
            "$date": "2020-02-05T13:15:14.000Z"
        },
        "rsvp_updated_at": {
            "$date": "2020-02-05T13:15:14.000Z"
        },
        "first_name": "James",
        "last_name": "Appleslice"}
  ]
}

So, the end-product would look something like this:

{'_id': ObjectId('61153253aBBBBBBBBBBBB'),
  'membershipNumber': 'N0XXXXXXXX',
  'an_user_id': '48XXXXXX',
  'n_email_clicks' : 412,
  'n_email_bounces' : 12,
  'n_rsvps' : 12
}

My idea was to use the $lookup parameter -- however, I only know how to use this for matching on fields that I have in the parent collection that I'm performing the aggregation on, but not on fields that have been generated in the process of the aggregation.

Any help would be hugely appreciated!

aggregation-framework pymongo mongodb
1 Answers

You could use $lookup pipeline. First you would $lookup the user id followed by another $lookup to verify if the user id exists in email. Lastly few more stages to collect the results and format per your need. Furthermore, you can add $out stage if you would like to write the results into another collection.

db.members.aggregate([{
  $lookup: {
    from: "an_users",
    let: {
      membershipNumber: "$membershipNumber"
    },
    pipeline: [
      {
        $match: {
          $expr: {
            $eq: [
              "$memref",
              "$$membershipNumber"
            ]
          },
          
        }
      },
      {
        "$lookup": {
          "from": "emails",
          "localField": "user_id",
          "foreignField": "click",
          "as": "clicks"
        }
      },
      {
        "$project": {
          "_id": 1,
          "membershipNumber": 1,
          "an_user_id": "$user_id",
          "n_email_clicks": {
            $size: "$clicks"
          }
        }
      }
    ],
    as: "details"
  }
},
{
  $replaceRoot: {
    newRoot: {
      $mergeObjects: [
        {
          $arrayElemAt: [
            "$details",
            0
          ]
        },
        "$$ROOT"
      ]
    }
  }
},
{
  $project: {
    details: 0
  }
}])

Working example - https://mongoplayground.net/p/yrFsNp44hpi

user_26838140
Related