Menu Docs

Página inicial do Docs → Desenvolver aplicações → Manual do MongoDB

start.process()

Nesta página

  • Definição
  • Sintaxe
  • Campos de comando
  • Comportamento
  • Controle de acesso
  • Exemplo
  • Saiba mais

Definição

sp.process()

Cria um processador defluxo efêmero na instância de processamento de fluxo atual.

Você só pode invocar este comando enquanto estiver conectado a uma instância de processamento de fluxo.

Este comando requer mongosh versão ≥ 2.0.

Sintaxe

O método sp.process() tem a seguinte sintaxe:

sp.process(
  [
    <pipeline>
  ],
  {
    <options>
  }
)

Campos de comando

sp.createStreamProcessor() usa estes campos:

Campo
Tipo
necessidade
Descrição
name
string
Obrigatório
Nome lógico para o processador de fluxo. Isso deve ser exclusivo dentro da instância do Atlas Stream Processing.
pipeline
array
Obrigatório
Transmita o pipeline de agregação que você deseja aplicar aos seus dados de streaming.
options
objeto
Opcional
objeto que define várias configurações opcionais para o processador de fluxo.
options.dlq
objeto
Condicional
Objeto que atribui uma dead letter queue (DLQ) para sua instância de processamento de fluxo. Este campo é necessário se você definir o campo options .
options.dlq.connectionName
string
Condicional
Etiqueta que identifica uma conexão em seu registro de conexão. Esta conexão deve fazer referência a um cluster do Atlas. Este campo é necessário se você definir o campo options.dlq .
options.dlq.db
string
Condicional
Nome de um reconhecimento de data center Atlas no cluster especificado em options.dlq.connectionName. Este campo é necessário se você definir o campo options.dlq .
options.dlq.coll
string
Condicional
Nome de uma collection no reconhecimento de data center especificado no options.dlq.db. Este campo é necessário se você definir o campo options.dlq .

Comportamento

sp.process() cria um processador de Atlas Stream Processing efêmero e sem nome na instância de Atlas Stream Processing atual e a inicializa imediatamente. Esse processador de Atlas Stream Processing só persiste enquanto for executado. Se você encerrar um processador de Atlas Stream Processing efêmero, deverá criá-lo novamente para usá-lo.

Controle de acesso

O usuário que executa sp.process() deve ter a função atlasAdmin .

Exemplo

O exemplo a seguir cria um processador de fluxo efêmero que ingere dados da conexão sample_stream_solar . O processador exclui todos os documentos onde o valor do campo device_id é device_8, passando o restante para uma janela em cascata com duração 10segundos. Cada janela agrupa os documentos que recebe e, em seguida, retorna várias estatísticas úteis de cada grupo. Em seguida, o processador de stream mescla esses registros com solar_db.solar_coll pela conexão mongodb1 .

sp.process(
  [
    {
      $source: {
        connectionName: 'sample_stream_solar',
        timeField: {
          $dateFromString: {
            dateString: '$timestamp'
          }
        }
      }
    },
    {
      $match: {
        $expr: {
          $ne: [
            "$device_id",
            "device_8"
          ]
        }
      }
    },
    {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        "pipeline": [
          {
            $group: {
              "_id": {  "device_id": "$device_id" },
              "max_temp": { $max: "$obs.temp" },
              "max_watts": { $max: "$obs.watts" },
              "min_watts": { $min: "$obs.watts" },
              "avg_watts": { $avg: "$obs.watts" },
              "median_watts": {
                                $median: {
                                  input: "$obs.watts",
                                  method: "approximate"
                                }
                              }
            }
          }
        ]
      }
    },
    {
      $merge: {
        into: {
          connectionName: "mongodb1",
          db: "solar_db",
          coll: "solar_coll"
        },
        on: ["_id"]
      }
    }
  ]
)

Saiba mais

← sp.listStreamProcessors()
start.processor.drop() →

Nesta página