«Выполнить агрегацию Монго» от NiFi, показывающую исключение JSONMappingException: нераспознанный токен ISODate

Question

Ниже приведены мои примерные данные: (Родительский JSON с информацией о этаже и на каждом этаже с информацией о нескольких датчиках)

{"Floor_Id": "Galileo_001",
    "name": "Forklifts",
    "Sensor_data": [{
            "Floor_Id": "Galileo_001",
            "Floor_name": "Forklifts",
            "Name": "forkLift_002",
            "Asset_Id": 123,
            "Load": 1.7096133,
            "Timestamp": 1537878750996
        },
        {
            "Floor_Id": "Galileo_001",
            "Floor_name": "Forklifts",
            "Name": "forkLift_003",
            "Asset_Id": 456,
            "Load": 1.7096133,
            "Timestamp": 1537878750996,
        },
        {
            "Floor_Id": "Galileo_001",
            "Floor_name": "Forklifts",
            "Name": "forkLift_005 ",
            "Asset_Id": 127,
            "Load": 1.7096133,
            "Timestamp": 1537878750996
        },
        {
            "Floor_Id": "Galileo_001",
            "Floor_name": "Forklifts",
            "Name": "forkLift_001",
            "Asset_Id": 157,
            "Load": 1.7096133,
            "Timestamp": 1537878750996,
        }
]}

Я хочу рассчитать общую нагрузку, общее расстояние и активные датчики для пола.И аналогичная информация (общая нагрузка и общее расстояние для каждого этажа) в день.Желаемый ответ выглядит следующим образом:

{
    "TotalLoad": 3214,
    "Active_no_of_sensor":5,
    "Floor_Id": "Galileo_001",
    "LoadUnit": "Kgs",
    "AssetStatus": [{
        "TotalLoad": 200,
        "LoadUnit": "Kgs",
        "Date": "1539588994"
    }, {
        "TotalLoad": 400,
        "LoadUnit": "Kgs",
        "Date": "1539475200"
    }, {
        "TotalLoad": 100,
        "LoadUnit": "Kgs",
        "Date": "1539388800"
    }]
}

В NiFi у меня есть два отдельных процессора агрегации Монго.

1. Для расчета общей нагрузки и общего количества активных датчиков для каждогоэтаж.
2. Для расчета общей нагрузки и т. Д. Для каждого этажа в день.

Для этого второго условия я использую следующий запрос:

[{"$unwind" : "$Sensor_data" },
{"$match": {"Floor_Id" : {"$eq": "${http.param.floorid}"}}},
{"$group": {
     "_id": {"Floor_Id": "$Floor_Id", 
         "DailyDate":{"$dateFromParts":{
                    "year":{"$year":{"$add": [ISODate("1970-01-01"), "$Sensor_data.Timestamp"]}},
                    "month":{"$month":{"$add": [ISODate("1970-01-01"), "$Sensor_data.Timestamp"]}},
                    "day":{"$dayOfMonth":{"$add": [ISODate("1970-01-01"), "$Sensor_data.Timestamp"]}}
                      } 
                  }
             },
     "AssetLoad": {"$sum": "$Sensor_data.Load" },
     "AssetDistance": {"$sum" : {"$subtract": [{"$toDecimal":"$Sensor_data.End_odometer"},
                                                {"$toDecimal":"$Sensor_data.Start_odometer"}]}}
    }
},
{
    "$group" : {
            "_id": {"Floor_Id": "$_id.Floor_Id"},
            "TotalLoad": { "$sum": "$AssetLoad" },
            "TotalDistance" : { "$sum": "$AssetDistance" },
            "AssetStatus":{
                "$push":{
                    "TotalLoad": "$AssetLoad",
                    "LoadUnit": "Kgs",
                    "TotalDistance" : "$AssetDistance",
                    "Date": "$_id.DailyDate"
                    }
                }
        }
},
{"$project" : {
    _id:0,
    "AssetStatus" : 1

    }
}]

Ошибка

Процессор показывает 'Error running mongodb aggregation query: Unrecognized Token ISODate: was expecting null, true, false or NaN'

Как можно решить эту проблему?(Я не хочу обновлять формат даты с BSON Long до ISODate в коллекции)

Answer 1

РАЗРЕШЕНО

Заменено ISODate на оператор $ convert в обработчике запуска Монго