Docs 菜单
Docs 主页
/
MongoDB Manual
/ / /

sp.createStreamProcessor()

在此页面上

  • 定义
  • 兼容性
  • 语法
  • 命令字段
  • 行为
  • 访问控制
  • 例子
  • 了解详情
sp.createStreamProcessor()

7.0版本新增: 在当前 流处理实例 上创建 流处理器 。

Atlas Stream Processing实例支持此方法。

sp.createStreamProcessor() 方法使用的语法如下:

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

sp.createStreamProcessor() 采用这些字段:

字段
类型
必要性
说明

name

字符串

必需

流处理器的逻辑名称。 它在Atlas Stream Processing实例中必须是唯一的。

pipeline

阵列

必需

要应用于流数据的流聚合管道

options

对象

Optional

为流处理器定义各种可选设置的对象。

options.dlq

对象

可选的

为您的 实例分配 死信队列Atlas Stream Processing 的对象。如果您定义options字段,则此字段是必需的。

options.dlq.connectionName

字符串

可选的

标识连接注册表中连接的标签。 此连接必须引用 Atlas 集群。 如果您定义options.dlq字段,则此字段是必需的。

options.dlq.db

字符串

可选的

options.dlq.connectionName中指定的集群上的 Atlas 数据库的名称。 如果您定义options.dlq字段,则此字段是必需的。

options.dlq.coll

字符串

可选的

options.dlq.db中指定的数据库中的集合名称。 如果您定义options.dlq字段,则此字段是必需的。

sp.createStreamProcessor() 在当前流处理实例上创建一个持久的、已命名的流处理器。 您可以使用 sp.processor.start()初始化此流处理器。 如果尝试创建与现有流处理器同名的流处理器, mongosh将返回错误。

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

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

sp.createStreamProcessor(
'solarDemo',
[
{
$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"]
}
}
]
)
  • 流聚合

  • 托管流处理器

后退

Atlas Stream Processing