Объединить после $ unwind для накопления временных рядов в Mongodb

Question

У меня есть коллекция документов временных рядов, которые следуют схеме:

{
    "_id" : ObjectId("5ce523fb3e9e92609c54747b"),
    "received" : ISODate("2018-06-01T00:00:00.000Z"),
    "payload" : {
        "tag1" : 0.0,
        "tag2" : 0.0,
        "tag3" : 0.0,
        ...
        "xyz": 0.0
    }
}

Полученные метки времени - ISO8601, а значения внутри полезной нагрузки - двойные.У меня нет контроля над схемой документов.Я получаю 1 документ в минуту, количество полей внутри полезной нагрузки во времени может варьироваться, и поэтому могут изменяться литералы ключей (имена тегов).

Мне, по сути, необходимо выполнять свертки времени (ежечасно, ежедневно,еженедельно и т. д.) показывает среднее значение для каждого данного тега за временной интервал.
После просмотра документации и связанных с ней сообщений (например, https://www.mongodb.com/blog/post/time-series-data-and-mongodb-part-3--querying-analyzing-and-presenting-timeseries-data), Я считаю, что это возможно.
Я считаю, что мне может понадобитьсясделайте что-то вроде unwind полезной нагрузки, а затем примените группировку агрегации по k, v, но это самое дальнее, что у меня есть:

db.my_data.aggregate([
    {"$project": {
        "year": {"$year": "$received"}, "month": {"$month": "$received"}, "dayOfMonth": {"$dayOfMonth": "$received"}, "hour": {"$hour": "$received"},
        "p": {"$objectToArray": "$payload"}}
    },
    {"$unwind": "$p"},
    {"$group": {
        _id: {           
            year: "$year",
            month: "$month",
            dayOfMonth: "$dayOfMonth",
            hour: "$hour",
            tag: "$p.k"
        },
        "t_avg": {$avg: "$p.v"},
        }
    },
])

Однако, в результате этого я получаю кучу «незаполненных»записей, столько же, сколько тегов внутри полезной нагрузки:

{
    "_id" : {
        "year" : 2018,
        "month" : 6,
        "dayOfMonth" : 1,
        "hour" : 0,
        "tag" : "tag1"
    },
    "t_avg" : 13.1261633627836
},
...

Это не то, что мне нужно. Очень важно, чтобы свернутые во времени записи были в том же формате, что и исходные, это: _id, received и payload, поэтому полученные средние значения для каждого тега по времени должны в конечном итоге объединиться в аналогичный объект полезной нагрузки

{ // assuming hour 5th
    "_id" : ObjectId("..."),
    "received" : ISODate("2018-06-01T00:05:00.000Z"),
    "payload" : {
        "tag1" : avg for the hour,
        "tag2" : avg for the hour,
        "tag3" : avg for the hour,
        ...
        "xyz": avg for the hour
    }
}

Я не знаю, как этого добитьсяthis.

Конечная цель состоит в том, чтобы создать представление с этим, чтобы свернутые средние значения могли быть получены по требованию без необходимости выполнять запрос как код из службы.Я не уверен, если синтаксис для создания представления отличается от синтаксиса агрегированного запроса, я считаю, что возможно создать представление на основе конвейера агрегации.

Answer 1

Вам необходимо преобразовать _id поля в дату (используя $ dateFromParts ) и использовать $ arrayToObject , чтобы получить payload, созданный из ваших динамически сгенерированных значений, попробуйте:

db.my_data.aggregate([
    {"$project": {
        "year": {"$year": "$received"}, "month": {"$month": "$received"}, "dayOfMonth": {"$dayOfMonth": "$received"}, "hour": {"$hour": "$received"},
        "p": {"$objectToArray": "$payload"}}
    },
    {"$unwind": "$p"},
    {"$group": {
        _id: {           
            year: "$year",
            month: "$month",
            dayOfMonth: "$dayOfMonth",
            hour: "$hour",
            tag: "$p.k"
        },
        "t_avg": {$avg: "$p.v"},
        }
    },
    {
        $group: {
            _id: { $dateFromParts: { year: "$_id.year", month: "$_id.month", day: "$_id.dayOfMonth", hour: "$_id.hour" } },
            payload: { $push: { k: "$_id.tag", v: "$t_avg" } }
        }
    },
    {
        $project: {
            _id: 1,
            payload: { $arrayToObject: "$payload" }
        }
    }
])