sp. 进程()（mongosh方法）

定义

sp.process()

在当前流处理工作区上创建临时流处理器。

兼容性

Atlas Stream Processing 工作区支持此方法。

语法

sp.process() 方法使用的语法如下：

sp.process(
  [
    <pipeline>
  ]
)

命令字段

sp.process() 采用这些字段：

字段	类型	必要性	说明
`pipeline`	阵列	必需	流处理器应用于流媒体数据的流聚合管道。
`options`	对象	Optional	为流处理器定义可选设置的对象。
`options.limit`	整型	Optional	返回到终端的最大文档数。 `sp.process()` 在返回达到此数量的文档后结束会话。
`options.tier`	字符串	Optional	要分配给流处理器的层。层名称不区分大小写。如果省略此字段，流处理器将使用流处理工作区的默认层级。必须是以下值之一： `SP2` `SP5` `SP10` `SP30` `SP50` 要学习；了解更多信息，请参阅层级。

行为

sp.process() 在当前流处理工作区上创建一个临时的未命名流处理器并立即将其初始化。此流处理器仅在运行时持续存在。如果终止临时流处理器，则必须重新创建才能使用。

发生以下任一情况时，临时流处理器会停止：

处理器返回 options.limit设立的文档数。
10 minutes elapse.
停止处理器。

访问控制

运行sp.process() 的用户必须具有 atlasAdmin角色。

例子

以下示例创建了一个临时流处理器，用于从sample_stream_solar 连接提取数据。处理器会排除device_id device_8字段值为10 的所有文档，将其余文档传递到持续时间为秒的滚动窗口。每个窗口对其接收的文档进行分组，然后返回群组的各种统计数据。然后，流处理器通过连接将这些记录合并到solar_db.solar_coll mongodb1。

sp.process(
  [
    {
      $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"]
      }
    }
  ]
)

了解详情

后退

sp.listWorkspaceDefaults

来年

sp.processor.drop