Definição
Novidades na versão 7.0: Cria um Processador de Stream no Espaço de Trabalho de Processamento de Streamatual.
Compatibilidade
Esse método é suportado em Atlas Stream Processing Workspaces.
Sintaxe
O método sp.createStreamProcessor() tem a seguinte sintaxe:
sp.createStreamProcessor( <name>, [ <pipeline> ], { <options> } )
Campos de comando
sp.createStreamProcessor() usa estes campos:
Campo | Tipo | necessidade | Descrição |
|---|---|---|---|
| string | Obrigatório | Nome lógico para o processador de stream. Isso deve ser exclusivo dentro do espaço de trabalho do stream processing. |
| array | Obrigatório | Transmita o pipeline de agregação que você deseja aplicar aos seus dados de streaming. |
| objeto | Opcional | objeto que define várias configurações opcionais para o processador de fluxo. ObservaçãoTodos os campos abaixo são opções que você pode configurar. |
| objeto | Condicional | Objeto que atribui uma fila de mensagens não entregues (DLQ) para seu espaço de trabalho de stream processing. Este campo é necessário se você definir o campo |
| string | Condicional | Etiqueta que identifica uma conexão em seu registro de conexão. Esta conexão deve fazer referência a um cluster Atlas. Este campo é necessário se você definir o campo |
| string | Condicional | Nome de um reconhecimento de data center Atlas no cluster especificado em |
| string | Condicional | Nome de uma collection no reconhecimento de data center especificado no |
| string | Opcional | O nível para o qual o Atlas Stream Processing atribui o processador. Se você não declarar esta opção, o Atlas Stream Processing atribuirá o processador ao nível do espaço de trabalho de stream processing. Deve ser um dos seguintes:
Para saber mais, consulte Níveis. |
Comportamento
sp.createStreamProcessor() cria um processador stream processing persistente e nomeado no espaço de trabalho stream processing atual. Você pode inicializar este processador de fluxo com sp.processor.start(). Se você tentar criar um processador de stream com o mesmo nome de um processador de stream existente, mongosh retornará um erro.
Controle de acesso
O usuário que executa o sp.createStreamProcessor() deve ter a função atlasAdmin.
Exemplo
O exemplo a seguir cria um processador de fluxo denominado solarDemo 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 uma 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 fluxo mescla esses registros com solar_db.solar_coll pela conexão mongodb1 .
sp.createStreamProcessor( 'solarDemo', [ { $source: { connectionName: 'sample_stream_solar', timeField: { $dateFromString: { dateString: '$timestamp' } } } }, { $match: { $expr: { $ne: [ "$device_id", "device_8" ] } } }, { $tumblingWindow: { interval: { size: Int32(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"] } } ] )