托管流处理器

先决条件

要创建和管理流处理器，您必须具备：

• Atlas Stream Processing实例

• mongosh 2.0或更高版本

• 具有atlasAdmin角色的数据库用户，用于创建和运行流处理器

• Atlas 集群

Considerations

许多流处理器命令要求您在方法调用中指定相关流处理器的名称。 以下部分中描述的语法假定严格为字母数字名称。 如果流处理器的名称包含非字母数字字符，例如连字符 ( - ) 或句点 ( . )，则必须将名称用方括号 ( [] ) 和双引号 ( "" ) 括在方法调用，如sp.["special-name-stream"].stats()中。

以交互方式创建流处理器

您可以使用sp.process()方法以交互方式创建流处理器。您以交互方式创建的流处理器表现出以下行为：

• 将输出和死信队列文档写入shell

• 创建后立即开始运行

• 运行10分钟或直到用户停止

• 停止后不要继续

您以交互方式创建的流处理器用于原型设计。要创建持久流处理器，请参阅创建流处理器。

sp.process() 通过以下语法实现：

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)

创建流处理器

要使用mongosh创建新的流处理器，请使用sp.createStreamProcessor()方法。 它具有以下语法：

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)

启动流处理器

要使用mongosh启动现有的流处理器，请使用sp.<streamprocessor>.start()方法。 <streamprocessor>必须是为当前流处理实例定义的流处理器名称。

例如，要启动名为proc01的流处理器，请运行以下命令：

sp.proc01.start()

此方法返回：

• true 如果流处理器存在且当前未运行。

• false 如果您尝试启动不存在或存在但当前正在运行的流处理器。

停止流处理器

要使用mongosh停止现有的流处理器，请使用sp.<streamprocessor>.stop()方法。 <streamprocessor>必须是为当前流处理实例定义的当前正在运行的流处理器的名称。

例如，要停止名为proc01的流处理器，请运行以下命令：

sp.proc01.stop()

此方法返回：

• true 如果流处理器存在且当前正在运行。

• false 如果流处理器不存在，或者流处理器当前未运行。

删除流处理器

要使用mongosh删除现有的流处理器，请使用sp.<streamprocessor>.drop()方法。 <streamprocessor>必须是为当前流处理实例定义的流处理器名称。

例如，要删除名为proc01的流处理器，请运行以下命令：

sp.proc01.drop()

此方法返回：

• true 如果流处理器存在。

• false 如果流处理器不存在。

删除流处理器时，Atlas Stream Processing 为其预配的所有资源以及所有已保存的状态都将被销毁。

列出可用的流处理器

要使用mongosh列出当前流处理实例上的所有可用流处理器，请使用sp.listStreamProcessors()方法。它返回一个文档列表，其中包含与每个流处理器关联的名称、开始时间、当前状态和管道。它具有以下语法：

sp.listStreamProcessors(<filter>)

<filter> 是一个文档，指定要按哪些字段筛选列表。

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
}

来自流处理器的样本

要使用mongosh从现有流处理器返回采样结果数组给STDOUT ，请使用sp.<streamprocessor>.sample()方法。 <streamprocessor>必须是为当前流处理实例定义的当前正在运行的流处理器的名称。例如，以下命令将从名为proc01的流处理器中进行采样。

sp.proc01.sample()

此命令会持续运行，直到您使用CTRL-C取消它，或者直到返回的样本大小累计达到40 MB。流处理器以以下形式的_dlqMessage文档报告样本中的无效文档：

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

您可以使用这些消息来诊断数据卫生问题，而无需定义死信队列集合。

查看流处理器的统计信息

要使用mongosh返回总结现有流处理器当前状态的文档，请使用sp.<streamprocessor>.stats()方法。 streamprocessor必须是为当前流处理实例定义的当前正在运行的流处理器的名称。它具有以下语法：

sp.<streamprocessor>.stats({options: {<options>}})

其中options是具有以下字段的可选文档：

输出文档包含以下字段：

例如，以下显示了名为inst01的Atlas Stream Processing实例上名为proc01的流处理器的状态，其中项大小以 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
  },
}