Agregación Acumular Objetos Internos

Como nota rápida, debe cambiar su "value" campo dentro de "values" ser numérico, ya que actualmente es una cadena. Pero vamos a la respuesta:

Si tiene acceso a $reduce desde MongoDB 3.4, entonces puedes hacer algo como esto:

db.collection.aggregate([
  { "$addFields": {
     "cities": {
       "$reduce": {
         "input": "$cities",
         "initialValue": [],
         "in": {
           "$cond": {
             "if": { "$ne": [{ "$indexOfArray": ["$$value._id", "$$this._id"] }, -1] },
             "then": {
               "$concatArrays": [
                 { "$filter": {
                   "input": "$$value",
                   "as": "v",
                   "cond": { "$ne": [ "$$this._id", "$$v._id" ] }
                 }},
                 [{
                   "_id": "$$this._id",
                   "name": "$$this.name",
                   "visited": {
                     "$add": [
                       { "$arrayElemAt": [
                         "$$value.visited",
                         { "$indexOfArray": [ "$$value._id", "$$this._id" ] }
                       ]},
                       1
                     ]
                   }
                 }]
               ]
             },
             "else": {
               "$concatArrays": [
                 "$$value",
                 [{
                   "_id": "$$this._id",
                   "name": "$$this.name",
                   "visited": 1
                 }]
               ]
             }
           }
         }
       }
     },
     "variables": {
       "$map": {
         "input": {
           "$filter": {
             "input": "$variables",
             "cond": { "$eq": ["$$this.name", "Budget"] } 
           }
         },
         "in": {
           "_id": "$$this._id",
           "name": "$$this.name",
           "defaultValue": "$$this.defaultValue",
           "lastValue": "$$this.lastValue",
           "value": { "$avg": "$$this.values.value" }
         }
       }
     }
  }}
])

Si tiene MongoDB 3.6, puede limpiarlo un poco con $mergeObjects :

db.collection.aggregate([
  { "$addFields": {
     "cities": {
       "$reduce": {
         "input": "$cities",
         "initialValue": [],
         "in": {
           "$cond": {
             "if": { "$ne": [{ "$indexOfArray": ["$$value._id", "$$this._id"] }, -1] },
             "then": {
               "$concatArrays": [
                 { "$filter": {
                   "input": "$$value",
                   "as": "v",
                   "cond": { "$ne": [ "$$this._id", "$$v._id" ] }
                 }},
                 [{
                   "_id": "$$this._id",
                   "name": "$$this.name",
                   "visited": {
                     "$add": [
                       { "$arrayElemAt": [
                         "$$value.visited",
                         { "$indexOfArray": [ "$$value._id", "$$this._id" ] }
                       ]},
                       1
                     ]
                   }
                 }]
               ]
             },
             "else": {
               "$concatArrays": [
                 "$$value",
                 [{
                   "_id": "$$this._id",
                   "name": "$$this.name",
                   "visited": 1
                 }]
               ]
             }
           }
         }
       }
     },
     "variables": {
       "$map": {
         "input": {
           "$filter": {
             "input": "$variables",
             "cond": { "$eq": ["$$this.name", "Budget"] } 
           }
         },
         "in": {
           "$mergeObjects": [
             "$$this",
             { "values": { "$avg": "$$this.values.value" } }
           ]
         }
       }
     }
  }}
])

Pero es más o menos lo mismo excepto que mantenemos los additionalData

Volviendo un poco antes de eso, siempre puede $unwind las "cities" acumular:

db.collection.aggregate([
  { "$unwind": "$cities" },
  { "$group": {
     "_id": { 
       "_id": "$_id",
       "cities": {
         "_id": "$cities._id",
         "name": "$cities.name"
       }
     },
     "_class": { "$first": "$class" },
     "name": { "$first": "$name" },
     "startTimestamp": { "$first": "$startTimestamp" },
     "endTimestamp" : { "$first": "$endTimestamp" },
     "source" : { "$first": "$source" },
     "variables": { "$first": "$variables" },
     "visited": { "$sum": 1 }
  }},
  { "$group": {
     "_id": "$_id._id",
     "_class": { "$first": "$class" },
     "name": { "$first": "$name" },
     "startTimestamp": { "$first": "$startTimestamp" },
     "endTimestamp" : { "$first": "$endTimestamp" },
     "source" : { "$first": "$source" },
     "cities": {
       "$push": {
         "_id": "$_id.cities._id",
         "name": "$_id.cities.name",
         "visited": "$visited"
       }
     },
     "variables": { "$first": "$variables" },
  }},
  { "$addFields": {
     "variables": {
       "$map": {
         "input": {
           "$filter": {
             "input": "$variables",
             "cond": { "$eq": ["$$this.name", "Budget"] } 
           }
         },
         "in": {
           "_id": "$$this._id",
           "name": "$$this.name",
           "defaultValue": "$$this.defaultValue",
           "lastValue": "$$this.lastValue",
           "value": { "$avg": "$$this.values.value" }
         }
       }
     }
  }}
])

Todos devuelven (casi) lo mismo:

{
        "_id" : ObjectId("5afc2f06e1da131c9802071e"),
        "_class" : "Traveler",
        "name" : "John Due",
        "startTimestamp" : 1526476550933,
        "endTimestamp" : 1526476554823,
        "source" : "istanbul",
        "cities" : [
                {
                        "_id" : "ef8f6b26328f-0663202f94faeaeb-1122",
                        "name" : "Cairo",
                        "visited" : 1
                },
                {
                        "_id" : "ef8f6b26328f-0663202f94faeaeb-3981",
                        "name" : "Moscow",
                        "visited" : 2
                }
        ],
        "variables" : [
                {
                        "_id" : "c8103687c1c8-97d749e349d785c8-9154",
                        "name" : "Budget",
                        "defaultValue" : "",
                        "lastValue" : "",
                        "value" : 3000
                }
        ]
}

Los dos primeros formularios son, por supuesto, lo mejor que se puede hacer, ya que simplemente trabajan "dentro" del mismo documento en todo momento.

Operadores como $reduce permite expresiones de "acumulación" en matrices, por lo que podemos usarlas aquí para mantener una matriz "reducida" que probamos para el único "_id" valor usando $indexOfArray para ver si ya hay un elemento acumulado que coincida. Un resultado de -1 significa que no está allí.

Para construir una "matriz reducida" tomamos el "initialValue" de [] como una matriz vacía y luego agréguela a través de $concatArrays . Todo ese proceso se decide a través del "ternario" $cond operador que considera el "if" condición y "then" o bien "se une" a la salida de $filter en el $$value actual para excluir el índice actual _id entrada, por supuesto con otra "matriz" que representa el objeto singular.

Para ese "objeto" usamos de nuevo el $indexOfArray para obtener realmente el índice coincidente ya que sabemos que el elemento "está ahí", y usarlo para extraer el "visited" actual valor de esa entrada a través de $arrayElemAt y $add a él para incrementar.

En el "else" caso, simplemente agregamos una "matriz" como un "objeto" que solo tiene un "visited" predeterminado valor de 1 . El uso de ambos casos acumula efectivamente valores únicos dentro de la matriz para generar.

En la última versión, simplemente $unwind la matriz y use los sucesivos $group etapas para "contar" primero con las entradas internas únicas y luego "reconstruir la matriz" en una forma similar.

Usando $unwind parece mucho más simple, pero dado que lo que realmente hace es tomar una copia del documento para cada entrada de la matriz, esto en realidad agrega una sobrecarga considerable al procesamiento. En las versiones modernas, generalmente hay operadores de matriz, lo que significa que no necesita usar esto a menos que su intención sea "acumular entre documentos". Entonces, si realmente necesita $group en un valor de una clave desde "dentro" de una matriz, entonces ahí es donde realmente necesita usarla.

En cuanto a las "variables" entonces simplemente podemos usar el $filter de nuevo aquí para obtener el "Budget" coincidente entrada. Hacemos esto como la entrada al $map operador que permite "remodelar" el contenido de la matriz. Principalmente queremos eso para que pueda tomar el contenido de los "values" (una vez que lo haga todo numérico) y use el $avg operador, que recibe esa forma de "notación de ruta de campo" directamente a los valores de la matriz porque, de hecho, puede devolver un resultado de dicha entrada.

Eso generalmente hace que el recorrido de casi TODOS los "operadores de matriz" principales para la canalización de agregación (excluyendo los operadores "conjunto") se encuentre dentro de una sola etapa de canalización.

Además, nunca olvides que casi siempre quieres $match con operadores de consulta regulares como la "primera etapa" de cualquier tubería de agregación para seleccionar los documentos que necesita. Idealmente usando un índice.

Suplentes

Los suplentes están trabajando en los documentos en código de cliente. Por lo general, no se recomienda, ya que todos los métodos anteriores muestran que en realidad "reducen" el contenido que devuelve el servidor, como suele ser el objetivo de las "agregaciones de servidores".

"Puede" ser posible debido a la naturaleza "basada en documentos" que los conjuntos de resultados más grandes pueden tomar mucho más tiempo usando $unwind y el procesamiento del cliente podría ser una opción, pero lo consideraría mucho más probable

A continuación, se muestra una lista que muestra la aplicación de una transformación a la secuencia del cursor a medida que se devuelven los resultados haciendo lo mismo. Hay tres versiones demostradas de la transformación, que muestran "exactamente" la misma lógica que la anterior, una implementación con lodash métodos de acumulación y una acumulación "natural" en el Map implementación:

const { MongoClient } = require('mongodb');
const { chain } = require('lodash');

const uri = 'mongodb://localhost:27017';
const opts = { useNewUrlParser: true };

const log = data => console.log(JSON.stringify(data, undefined, 2));

const transform = ({ cities, variables, ...d }) => ({
  ...d,
  cities: cities.reduce((o,{ _id, name }) =>
    (o.map(i => i._id).indexOf(_id) != -1)
      ? [
          ...o.filter(i => i._id != _id),
          { _id, name, visited: o.find(e => e._id === _id).visited + 1 }
        ]
      : [ ...o, { _id, name, visited: 1 } ]
  , []).sort((a,b) => b.visited - a.visited),
  variables: variables.filter(v => v.name === "Budget")
    .map(({ values, additionalData, ...v }) => ({
      ...v,
      values: (values != undefined)
        ? values.reduce((o,e) => o + e.value, 0) / values.length
        : 0
    }))
});

const alternate = ({ cities, variables, ...d }) => ({
  ...d,
  cities: chain(cities)
    .groupBy("_id")
    .toPairs()
    .map(([k,v]) =>
      ({
        ...v.reduce((o,{ _id, name }) => ({ ...o, _id, name }),{}),
        visited: v.length
      })
    )
    .sort((a,b) => b.visited - a.visited)
    .value(),
  variables: variables.filter(v => v.name === "Budget")
    .map(({ values, additionalData, ...v }) => ({
      ...v,
      values: (values != undefined)
        ? values.reduce((o,e) => o + e.value, 0) / values.length
        : 0
    }))

});

const natural = ({ cities, variables, ...d }) => ({
  ...d,
  cities: [
    ...cities
      .reduce((o,{ _id, name }) => o.set(_id,
        [ ...(o.has(_id) ? o.get(_id) : []), { _id, name } ]), new Map())
      .entries()
  ]
  .map(([k,v]) =>
    ({
      ...v.reduce((o,{ _id, name }) => ({ ...o, _id, name }),{}),
      visited: v.length
    })
  )
  .sort((a,b) => b.visited - a.visited),
  variables: variables.filter(v => v.name === "Budget")
    .map(({ values, additionalData, ...v }) => ({
      ...v,
      values: (values != undefined)
        ? values.reduce((o,e) => o + e.value, 0) / values.length
        : 0
    }))

});

(async function() {

  try {

    const client = await MongoClient.connect(uri, opts);

    let db = client.db('test');
    let coll = db.collection('junk');

    let cursor = coll.find().map(natural);

    while (await cursor.hasNext()) {
      let doc = await cursor.next();
      log(doc);
    }

    client.close();

  } catch(e) {
    console.error(e)
  } finally {
    process.exit()
  }

})()