managed processadores de fluxo

Pré-requisitos

Para criar e managed um processador de stream, você deve ter:

• Uma instância de Atlas Stream Processing

• mongosh versão 2.0 ou superior

• Um utilizador de banco de dados com a função atlasAdmin para criar e executar processadores de stream

• Um cluster do Atlas

Considerações

Muitos comandos do processador de fluxo exigem que você especifique o nome do processador de fluxo relevante na invocação do método. A sintaxe descrita nas seções a seguir assume nomes estritamente alfanuméricos. Se o nome do processador de stream incluir caracteres não alfanuméricos, como hífens (-) ou pontos finais (.), você deverá colocar o nome entre colchetes ([]) e aspas duplas ("") no invocação de método, como em sp.["special-name-stream"].stats().

Crie um processador de stream interativamente

Você pode criar um processador de stream interativamente com o método sp.process() . Os processadores de fluxo que você cria interativamente exibem o seguinte comportamento:

• Gravar documentos de saída e dead letter queue (DLQ) no shell

• Começam a ser executados imediatamente após a criação

• Execute por 10 minutos ou até que o usuário pare

• Não persista depois de parar

Os processadores de fluxo que você cria interativamente destinam-se à criação de protótipos. Para criar um processador de fluxo persistente, consulte Criar um processador de fluxo.

sp.process() tem a seguinte sintaxe:

sp.process(<pipeline>)

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",
   }
 }
]

sp.process(pipeline)

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)

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 stream 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.

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 atualmente em execução 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 stream não existir ou se o processador de stream não estiver em execução no momento.

Descartar 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.

Liste os processadores de fluxo disponíveis

Para listar todos os processadores de fluxo disponíveis na instância de processamento de fluxo atual com mongosh, utilize o método sp.listStreamProcessors() . Ele retorna uma lista de documentos que contêm o nome, hora de início, estado atual e pipeline associados a cada processador de stream. 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
}

Amostra de um processador de fluxo

Para retornar uma array de resultados de amostra de um processador de stream existente para STDOUT com mongosh, use o método sp.<streamprocessor>.sample() . <streamprocessor> deve ser o nome de um processador de fluxo atualmente em execução definido para a instância de processamento de fluxo atual. Por exemplo, as seguintes amostras de comando de um processador de stream denominado proc01.

sp.proc01.sample()

Esse comando é executado continuamente até que você o cancele usando CTRL-C ou até que as amostras retornadas atinjam cumulativamente 40 MB de tamanho. O processador de fluxo relata documentos inválidos na amostra em um documento _dlqMessage do seguinte formato:

 {
   _dlqMessage: {
     _stream_meta: {
       source: {
         type: "<type>"
       }
     },
     errInfo: {
       reason: "<reasonForError>"
     },
     doc: {
       _id: ObjectId('<group-id>'),
       ...
     },
     processorName: '<procName>'
   }
}

Você pode usar essas mensagens para diagnosticar problemas de limpeza de dados sem definir uma coleção de fila de letras mortas .

Ver estatísticas de 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 atualmente em execução 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
  },
}