Guia de configuração do Atlas Stream Processing para usuários do Kafka Connector

1
{
2
    "name": "mongo-tutorial-sink",
3
    "config": {
4
      "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
5
      "topics": "Stocks",
6
      "connection.uri": "mongodb://mongo1",
7
      "key.converter": "org.apache.kafka.connect.storage.StringConverter",
8
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
9
      "value.converter.schemas.enable": false,
10
      "database": "Stocks",
11
      "collection": "StockData",
12
      "mongo.errors.tolerance":"data",
13
      "errors.tolerance":"all",
14
      "errors.deadletterqueue.topic.name":"mongoerror",
15
      "errors.deadletterqueue.topic.replication.factor":1
16
      }
17
  }

1
{
2
"exchange":"NASDAQ"
3
"company_symbol":"AHL"
4
"company_name":"AMUSED HAWK LABS"
5
"price":8.41
6
"tx_time":"2024-04-03T14:51:40Z"
7
}

1
s_kafka = { $source: { connectionName: "HostedKafka", "topic": "Stocks" } }

1
tw = { $tumblingWindow: { interval: { size: 5, unit: "second" }, pipeline: [{ $group: { _id: "$company_symbol", max: { $max: "$price" }, avg: { $avg: "$price" } } }] } }

1
sp.process([s_kafka,tw])

1
m={$merge: {  into: { connectionName: "AtlasSource",db: "Stock",coll: "StockReport" }}}

1
sp.createStreamProcessor(‘mytestsp’,[s_kafka,tw,m])

1
mytestsp.start()

1
mytestsp.stats()

1
{
2
  ok: 1,
3
  ns: '66744a49d8619c0695a7e5e9.5e98cd46cd7309236ceaab3c.testsp',
4
  stats: {
5
    name: 'mytestsp',
6
    processorId: '66780d901f7c0b19f89f559f',
7
    status: 'running',
8
    scaleFactor: Long('1'),
9
    inputMessageCount: Long('633972'),
10
    inputMessageSize: 280913520,
11
    outputMessageCount: Long('296963'),
12
    outputMessageSize: 475437763,
13
    dlqMessageCount: Long('0'),
14
    dlqMessageSize: 0,
15
    stateSize: 3328,
16
    memoryTrackerBytes: 33554432,
17
    watermark: ISODate('2024-08-07T20:49:43.047Z'),
18
    ok: 1
19
  },
20
  pipeline: [
21
    {
22
      '$source': {
23
        connectionName: ‘HostedKafka’,
24
        Topic: ‘Stocks’
25
      }
26
    },
27
    {
28
      '$tumblingWindow': {
29
        interval: { size: 5, unit: 'second' },
30
        pipeline: [
31
          {
32
            '$group': {
33
              _id: [Object],
34
              max: [Object],
35
              avg: [Object]
36
            }
37
          }
38
        ]
39
      }
40
    },
41
    {
42
      '$merge': {
43
        into: {
44
          connectionName: 'AtlasSource',
45
          db: 'Stock',
46
          coll: 'StockReport'
47
        }
48
      }
49
    }
50
  ]
51
}

1
{
2
  device_id: 'device_0',
3
  group_id: 8,
4
  timestamp: '2024-04-04T14:49:52.943+00:00',
5
  max_watts: 250,
6
  event_type: 0,
7
  obs: {
8
    watts: 75,
9
    temp: 5
10
  },
11
  _ts: ISODate('2024-04-04T14:49:52.943Z')
12
}

1
post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
2
value.projection.type=AllowList
3
value.projection.list=device_id,obs.watts,obs.temp

1
 {
2
  device_id: 'device_0',
3
  obs: {
4
    watts: 75,
5
    temp: 5
6
  }
7
}

1
{$project:{"device_id":1,"obs.watts":1, "obs.temp":1}}

1
 {
2
  device_id: 'device_1',
3
  obs: {
4
    Watts: 50,
5
    temp: 10
6
  }
7
}