Explorando operadores de janela no processamento de fluxo Atlas

1
AtlasStreamProcessing> sp.listConnections()
2
{
3
    ok: 1,
4
    connections: [
5
    	{
6
            name: 'sample_stream_solar',
7
            type: 'inmemory',
8
            createdAt: ISODate("2023-08-26T18:42:48.357Z")
9
        } 
10
    ]
11
}

1
var solarstream={ $source: { "connectionName": "sample_stream_solar" } }

1
sp.process([solarstream])

1
{
2
    device_id: 'device_2',
3
    group_id: 3,
4
    timestamp: '2023-08-27T13:51:53.375+00:00',
5
    max_watts: 250,
6
    event_type: 0,
7
    obs: {
8
        watts: 168,
9
        temp: 15
10
    },
11
    _ts: ISODate("2023-08-27T13:51:53.375Z"),
12
    _stream_meta: {
13
        sourceType: 'sampleData',
14
        timestamp: ISODate("2023-08-27T13:51:53.375Z")
15
    }
16
}

1
var Twindow= { 
2
    $tumblingWindow: { 
3
         interval: { size: NumberInt(1), unit: "minute" },
4
         pipeline: [ 
5
             { 
6
                 $group: { 	
7
                     _id: "$device_id", 
8
                    max: { $max: "$obs.watts" }, 
9
                    avg: { $avg: "$obs.watts" } 
10
                 }
11
             }
12
         ]
13
     } 
14
}

1
sp.process([solarstream,Twindow])

1
{
2
    _id: 'device_4',
3
    max: 236,
4
    avg: 95,
5
    _stream_meta: {
6
        sourceType: 'sampleData',
7
        windowStartTimestamp: ISODate("2023-08-27T13:59:00.000Z"),
8
        windowEndTimestamp: ISODate("2023-08-27T14:00:00.000Z")
9
    }
10
}
11
{
12
    _id: 'device_2',
13
    max: 211,
14
    avg: 117.25,
15
    _stream_meta: {
16
        sourceType: 'sampleData',
17
        windowStartTimestamp: ISODate("2023-08-27T13:59:00.000Z"),
18
        windowEndTimestamp: ISODate("2023-08-27T14:00:00.000Z")
19
    }
20
}

1
var m= { '$match': { device_id: 'device_1' } }

1
var Twindow= {
2
    '$tumblingWindow': {
3
        interval: { size: Int32(1), unit: 'minute' },
4
        pipeline: [
5
            {
6
                '$group': {
7
                    _id: '$device_id',
8
                    max: { '$max': '$obs.watts' },
9
                    avg: { '$avg': '$obs.watts' }
10
                }
11
            }
12
        ]
13
    }
14
}
15
16
var delta = { '$addFields': { delta: { '$subtract': ['$max', '$avg'] } } }

1
sp.process([solarstream,m,Twindow,delta])

1
{
2
    _id: 'device_1',
3
    max: 238,
4
    avg: 75.3,
5
    _stream_meta: {
6
        sourceType: 'sampleData',
7
        windowStartTimestamp: ISODate("2023-08-27T19:11:00.000Z"),
8
        windowEndTimestamp: ISODate("2023-08-27T19:12:00.000Z")
9
    },
10
    delta: 162.7
11
}
12
{
13
    _id: 'device_1',
14
    max: 220,
15
    avg: 125.08333333333333,
16
    _stream_meta: {
17
        sourceType: 'sampleData',
18
        windowStartTimestamp: ISODate("2023-08-27T19:12:00.000Z"),
19
        windowEndTimestamp: ISODate("2023-08-27T19:13:00.000Z")
20
    },
21
    delta: 94.91666666666667
22
}
23
{
24
    _id: 'device_1',
25
    max: 238,
26
    avg: 119.91666666666667,
27
    _stream_meta: {
28
        sourceType: 'sampleData',
29
        windowStartTimestamp: ISODate("2023-08-27T19:13:00.000Z"),
30
        windowEndTimestamp: ISODate("2023-08-27T19:14:00.000Z")
31
    },
32
    delta: 118.08333333333333
33
}

1
var Hwindow = {
2
    '$hoppingWindow': {
3
        interval: { size: 1, unit: 'minute' },
4
        hopSize: { size: 30, unit: 'second' },
5
        pipeline: [
6
            {
7
                '$group': {
8
                    _id: '$device_id',
9
                    max: { '$max': '$obs.watts' },
10
                    avg: { '$avg': '$obs.watts' }
11
                }
12
            }
13
        ]
14
    }
15
}

1
var m = { '$match': { device_id: 'device_1' } }

1
sp.process([solarstream,m,Hwindow])

1
{
2
    _id: 'device_1',
3
    max: 238,
4
    avg: 76.625,
5
    _stream_meta: {
6
        sourceType: 'sampleData',
7
        windowStartTimestamp: ISODate("2023-08-27T19:37:30.000Z"),
8
        windowEndTimestamp: ISODate("2023-08-27T19:38:30.000Z")
9
    }
10
}
11
{
12
    _id: 'device_1',
13
    max: 238,
14
    avg: 82.71428571428571,
15
    _stream_meta: {
16
        sourceType: 'sampleData',
17
        windowStartTimestamp: ISODate("2023-08-27T19:38:00.000Z"),
18
        windowEndTimestamp: ISODate("2023-08-27T19:39:00.000Z")
19
    }
20
}
21
{
22
    _id: 'device_1',
23
    max: 220,
24
    avg: 105.54545454545455,
25
    _stream_meta: {
26
        sourceType: 'sampleData',
27
        windowStartTimestamp: ISODate("2023-08-27T19:38:30.000Z"),
28
        windowEndTimestamp: ISODate("2023-08-27T19:39:30.000Z")
29
    }
30
}