managed processadores de fluxo

A API de Administração do Atlas fornece um endpoint para criar um processador de fluxo.

Crie um processador de fluxo

Para criar um novo processador de fluxo com mongosh, use o método sp.createStreamProcessor() . Tem a seguinte sintaxe:

sp.createStreamProcessor(<name>, <pipeline>, <options>)

mongosh "mongodb://atlas-stream-78xq9gi1v4b4288x06a73e9f-zi30g.virginia-usa.a.query.mongodb-qa.net/?authSource=%24external&authMechanism=MONGODB-X509" \\
--tls --authenticationDatabase admin --username streamOwner

pipeline = [
  {$source: {"connectionName": "myKafka", "topic": "stuff"}},
  {$match: { temperature: 46 }},
  {
    "$emit": {
      "connectionName": "mySink",
      "topic" : "output",
    }
  }
]

deadLetter = {
  dlq: {
    connectionName: "cluster01",
    db: "metadata",
    coll: "dlq"
  }
}

sp.createStreamProcessor("proc01", pipeline, deadLetter)

A API Atlas Administration fornece um endpoint para iniciar um processador de fluxo.

Inicie um processador de fluxo

Para iniciar um processador de fluxo existente com mongosh, use o método sp.<streamprocessor>.start(). <streamprocessor> deve ser o nome de um processador de fluxo definido para a instância de processamento de fluxo atual.

Por exemplo, para iniciar um processador de fluxo denominado proc01, execute o seguinte comando:

sp.proc01.start()

Este método retorna:

• true se o processador de fluxo existir e não estiver em execução no momento.

• false se você tentar iniciar um processador de fluxo que não existe ou existe e está em execução no momento.

A API de Administração do Atlas fornece um endpoint para interromper um processador de fluxo.

Interromper um processador de fluxo

Para interromper um processador de fluxo existente com mongosh, use o método sp.<streamprocessor>.stop(). <streamprocessor> deve ser o nome de um processador de fluxo em execução no momento definido para a instância de processamento de fluxo atual.

Por exemplo, para parar um processador de stream denominado proc01, execute o seguinte comando:

sp.proc01.stop()

Este método retorna:

• true se o processador de stream existir e estiver em execução.

• false se o processador de fluxo não existir ou se o processador de fluxo não estiver em execução no momento.

A API de Administração do Atlas fornece um endpoint para excluir um processador de fluxo.

Excluir um processador de fluxo

Para excluir um processador de fluxo existente com mongosh, use o método sp.<streamprocessor>.drop(). <streamprocessor> deve ser o nome de um processador de fluxo definido para a instância de processamento de fluxo atual.

Por exemplo, para eliminar um processador de fluxo denominado proc01, execute o seguinte comando:

sp.proc01.drop()

Este método retorna:

• true se o processador de fluxo existir.

• false se o processador de fluxo não existir.

Quando você descarta um processador de fluxo, todos os recursos que o Atlas Stream Processing provisionou para ele são destruídos, junto com todo o estado salvo.

A API de Administração do Atlas fornece um endpoint para listar todos os processadores de fluxo disponíveis.

Excluir um processador de fluxo

Para listar todos os processadores de fluxo disponíveis na instância de processamento de fluxo atual com mongosh, use o método sp.listStreamProcessors(). Ele retorna uma lista de documentos contendo o nome, a hora de início, o estado atual e o respectivo pipeline de cada processador de fluxo. Tem a seguinte sintaxe:

sp.listStreamProcessors(<filter>)

<filter> é um documento que especifica por quais campos filtrar a lista.

sp.listStreamProcessors()

1
{
2
  id: '0135',
3
  name: "proc01",
4
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "RUNNING",
6
  error_msg: '',
7
  pipeline: [
8
    {
9
      $source: {
10
        connectionName: "myKafka",
11
        topic: "stuff"
12
      }
13
    },
14
    {
15
      $match: {
16
        temperature: 46
17
      }
18
    },
19
    {
20
      $emit: {
21
        connectionName: "mySink",
22
        topic: "output",
23
      }
24
    }
25
  ],
26
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
27
},
28
{
29
  id: '0218',
30
  name: "proc02",
31
  last_modified: ISODate("2023-03-21T20:17:33.601Z"),
32
  state: "STOPPED",
33
  error_msg: '',
34
  pipeline: [
35
    {
36
      $source: {
37
        connectionName: "myKafka",
38
        topic: "things"
39
      }
40
    },
41
    {
42
      $match: {
43
        temperature: 41
44
      }
45
    },
46
    {
47
      $emit: {
48
        connectionName: "mySink",
49
        topic: "results",
50
      }
51
    }
52
  ],
53
  lastStateChange: ISODate("2023-03-21T20:18:26.139Z")
54
}

sp.listStreamProcessors({"state": "running"})

1
{
2
  id: '0135',
3
  name: "proc01",
4
  last_modified: ISODate("2023-03-20T20:15:54.601Z"),
5
  state: "RUNNING",
6
  error_msg: '',
7
  pipeline: [
8
    {
9
      $source: {
10
        connectionName: "myKafka",
11
        topic: "stuff"
12
      }
13
    },
14
    {
15
      $match: {
16
        temperature: 46
17
      }
18
    },
19
    {
20
      $emit: {
21
        connectionName: "mySink",
22
        topic: "output",
23
      }
24
    }
25
  ],
26
  lastStateChange: ISODate("2023-03-20T20:15:59.442Z")
27
}

A API de Administração do Atlas fornece um endpoint para visualizar as estatísticas de um processador de fluxo.

Obtenha um processador de fluxo

Para retornar um documento resumindo o status atual de um processador de fluxo existente com mongosh, use o método sp.<streamprocessor>.stats(). streamprocessor deve ser o nome de um processador de fluxo em execução no momento definido para a instância de processamento de fluxo atual. Tem a seguinte sintaxe:

sp.<streamprocessor>.stats({options: {<options>}})

Onde options é um documento opcional com os seguintes campos:

O documento de saída tem os seguintes campos:

Por exemplo, o seguinte mostra o status de um processador de Atlas Stream Processing denominado proc01 em uma instância de Atlas Stream Processing denominada inst01 com tamanhos de item exibidos em KB:

sp.proc01.stats(1024)
{
  ok: 1,
  ns: 'inst01',
  stats: {
    name: 'proc01',
    status: 'running',
    scaleFactor: Long("1"),
    inputMessageCount: Long("706028"),
    inputMessageSize: 958685236,
    outputMessageCount: Long("46322"),
    outputMessageSize: 85666332,
    dlqMessageCount: Long("0"),
    dlqMessageSize: Long("0"),
    stateSize: Long("2747968"),
    watermark: ISODate("2023-12-14T14:35:32.417Z"),
    ok: 1
  },
}