Introduction to Atlas Stream Processing Development

1
// our array of stages
2
// source is defined earlier
3
sp_aggregation_pipeline = [source]
4
sp.createStreamProcessor("SP_NAME", sp_aggregation_pipeline, <OPTIONS>)

1
sp_aggregation_pipeline = [source, stage_1, stage_2...]
2
sp.createStreamProcessor("SP_NAME", sp_aggregation_pipeline, <OPTIONS>)

1
src_hard_coded  = {
2
  $source: {
3
    // our hard-coded dataset
4
    documents: [
5
      {'id':  'entity_1', 'value':  1},
6
      {'id':  'entity_1', 'value':  3},
7
      {'id':  'entity_2', 'value':  7},
8
      {'id':  'entity_1', 'value':  4},
9
      {'id':  'entity_2', 'value':  1}
10
     ]
11
    }
12
  }
13
sp.process( [src_hard_coded] );

1
{
2
  id: 'entity_1',
3
  value: 1,
4
  _ts: 2024-02-14T18:52:33.704Z,
5
  _stream_meta: { timestamp: 2024-02-14T18:52:33.704Z }
6
}
7
{
8
  id: 'entity_1',
9
  value: 3,
10
  _ts: 2024-02-14T18:52:33.704Z,
11
  _stream_meta: { timestamp: 2024-02-14T18:52:33.704Z }
12
}
13
...

1
src_hard_coded  = {
2
  $source: {
3
    // our hard-coded dataset
4
    documents: [
5
      {'id':  'entity_1', 'value':  1},
6
      {'id':  'entity_1', 'value':  3},
7
      {'id':  'entity_2', 'value':  7},
8
      {'id':  'entity_1', 'value':  4},
9
      {'id':  'entity_2', 'value':  1}
10
     ]
11
    }
12
  }
13
14
w  = {
15
  $tumblingWindow: {
16
    // This is the slice of time we want to look at every iteration
17
    interval: {size:  NumberInt(2), unit:  "second"},
18
    // If no additional data is coming in, idleTimeout defines when the window is forced to close
19
    idleTimeout : {size:  NumberInt(2), unit:  "second"},
20
    "pipeline": [
21
      {
22
        '$group': {
23
            '_id':  '$id',
24
            'values': { '$push':  "$value" }
25
          }
26
        }
27
      ]
28
    }
29
  }
30
sp_pipeline = [src_hard_coded, w];
31
sp.process( sp_pipeline );

1
{
2
  _id: 'entity_2',
3
  values: [ 7, 1 ],
4
  _stream_meta: {
5
    windowStartTimestamp: 2024-02-14T19:29:46.000Z,
6
    windowEndTimestamp: 2024-02-14T19:29:48.000Z
7
  }
8
}
9
{
10
  _id: 'entity_1',
11
  values: [ 1, 3, 4 ],
12
  _stream_meta: {
13
    windowStartTimestamp: 2024-02-14T19:29:46.000Z,
14
    windowEndTimestamp: 2024-02-14T19:29:48.000Z
15
  }
16
}

1
...
2
$tumblingWindow: {
3
    interval: {size:  NumberInt(10), unit:  "second"},
4
    idleTimeout : {size:  NumberInt(10), unit:  "second"},
5
...

1
{
2
  _id: 'entity_1',
3
  values: [ 1 ],
4
  _stream_meta: {
5
    windowStartTimestamp: 2024-02-13T14:51:30.000Z,
6
    windowEndTimestamp: 2024-02-13T14:51:40.000Z
7
  }
8
}
9
{
10
  _id: 'entity_1',
11
  values: [ 3, 4 ],
12
  _stream_meta: {
13
    windowStartTimestamp: 2024-02-13T14:51:40.000Z,
14
    windowEndTimestamp: 2024-02-13T14:51:50.000Z
15
  }
16
}
17
{
18
  _id: 'entity_2',
19
  values: [ 7, 1 ],
20
  _stream_meta: {
21
    windowStartTimestamp: 2024-02-13T14:51:40.000Z,
22
    windowEndTimestamp: 2024-02-13T14:51:50.000Z
23
  }
24
}

1
src_hard_coded  = {
2
  $source: {
3
    // define our event "timestamp_gps" as the _ts
4
    timeField: { '$dateFromString': { dateString:  '$timestamp_msg' } },
5
    // our hard-coded dataset
6
    documents: [
7
      {'id':  'entity_1', 'value':  1, 'timestamp_msg':  '2024-02-13T14:51:39.402336'},
8
      {'id':  'entity_1', 'value':  3, 'timestamp_msg':  '2024-02-13T14:51:41.402674'},
9
      {'id':  'entity_2', 'value':  7, 'timestamp_msg':  '2024-02-13T14:51:43.402933'},
10
      {'id':  'entity_1', 'value':  4, 'timestamp_msg':  '2024-02-13T14:51:45.403352'},
11
      {'id':  'entity_2', 'value':  1, 'timestamp_msg':  '2024-02-13T14:51:47.403752'}
12
     ]
13
    }
14
  }