开始使用 Atlas Stream Processing

要在Atlas用户界面中创建流处理器，Go转到Atlas项目的 Stream Processing 页面，然后单击流处理实例窗格中的 Configure。 然后选择使用可视化生成器或JSON编辑器来配置名为 solarDemo 的流处理器：

在 mongosh 中运行以下命令以创建名为 solarDemo 的持久流处理器：

1. 连接到您的Atlas Stream Processing实例。

   使用与您的 实例关联的连接stringAtlas Stream Processing mongosh，通过 进行连接。

   1. 在Atlas Stream Processing实例的窗格中，单击 Connect。

   2. 在 Connect to your instance 对话框中，选择 Shell标签页。

   3. 复制对话框中显示的连接字符串。 它采用以下格式，其中 <atlas-stream-processing-url> 是流处理实例的URL ，<username> 是具有 atlasAdmin 角色的数据库用户的用户名：

      mongosh "mongodb://<atlas-stream-processing-url>/"
      --tls --authenticationDatabase admin --username <username>
      --password <password>

   4. 将连接字符串粘贴到终端中，并将 <password> 占位符替换为用户的凭证。

      按 Enter运行该程序并连接到您的流处理实例。

2. 配置 $source 阶段。

   为从 sample_stream_solar 源摄取数据的 $source 阶段定义一个变量。

   let s = {
     $source: {
       connectionName: "sample_stream_solar"
     }
   }

3. 配置 $group 阶段。

   为 $group 阶段定义一个变量，该变量根据其 group_id 导出每个太阳能设备的最高温度以及平均瓦数、最大瓦数和最小瓦数。

   let g = {
     $group: {
       _id: "$group_id",
       max_temp: {
         $max: "$obs.temp"
       },
       avg_watts: {
          $avg: "$obs.watts"
       },
       max_watts: {
         $max: "$obs.watts"
       },
       min_watts: {
         $min: "$obs.watts"
       }
     }
   }

4. 配置 $tumblingWindow 阶段。

   为了流媒体数据执行 $group 等累加操作， Atlas Stream Processing使用Windows来绑定数据集。为 $tumblingWindow 阶段定义一个变量，将流分成连续的 10 秒间隔。

   这意味着，示例，当 $group 阶段计算 max_watts 的值时，它会从前 10 秒内摄取的具有给定 group_id 的所有文档的 obs.watts 值中提取最大值。

    let t = {
      $tumblingWindow: {
        interval: {
          size: NumberInt(10),
          unit: "second"
        },
        pipeline: [g]
     }
   }

5. 配置 $merge 阶段。

   为 $merge 阶段定义一个变量，该阶段将处理后的流媒体数据写入已连接Atlas 集群的 solarDb数据库中名为 solarColl 的集合。

   let m = {
     $merge: {
       into: {
         connectionName: "mongodb1",
         db: "solarDb",
         coll: "solarColl"
       }
     }
   }

6. 创建流处理器。

   使用 sp.createStreamProcessor() 方法为新的流处理器指定名称并声明其聚合管道。$group 阶段属于 $tumblingWindow 的嵌套管道，不得将其包含在处理器管道定义中。

   sp.createStreamProcessor("solarDemo", [s, t, m])

   这将创建一个名为 solarDemo 的流处理器，该处理器应用之前定义的查询，并将处理后的数据写入您连接到的集群上 solarDb 数据库的 solarColl 集合。它返回从太阳能设备的 10 秒间隔观测得出的各种测量值。

   要详细学习；了解Atlas Stream Processing如何写入静态数据库，请参阅 $merge（流处理）。

在流处理实例的流处理器列表中，单击流处理器的 Start 图标。

使用 中的sp.processor.start() mongosh方法：

sp.solarDemo.start()

要验证流处理器是否正在将数据写入 Atlas 集群：

1. 在 Atlas 中，进入项目的 Clusters 页面。

   警告： 导航改进正在进行中

   我们目前正在推出改进的全新导航体验。如果以下步骤与Atlas用户界面中的视图不匹配，请参阅预览文档。

   1. 如果尚未显示，请从导航栏上的 Organizations 菜单中选择包含所需项目的组织。

   2. 如果尚未显示，请从导航栏的Projects菜单中选择所需的项目。

   3. 如果尚未出现，请单击侧边栏中的 Clusters（集群）。

      会显示集群页面。

2. 单击集群的对应 Browse Collections 按钮。

   显示数据浏览器。

3. 查看 MySolar 集合。

要验证处理器是否处于活动状态，请使用 中的sp.processor.stats() mongosh方法：

sp.solarDemo.stats()

此方法报告 solarDemo流处理器的操作统计信息。

您还可以使用sp.processor.sample() 中的mongosh 方法返回终端中已处理文档的样本。

sp.solarDemo.sample()

{
  _id: 10,
  max_temp: 16,
  avg_watts: 232,
  max_watts: 414,
  min_watts: 73
}

在流处理实例的流处理器列表中，单击流处理器的 Delete () 图标。

在出现的确认对话框中，键入流处理器的名称 (solarDemo) 以确认要将其删除，然后单击 Delete。

使用 中的sp.processor.drop() mongosh方法删除solarDemo ：

sp.solarDemo.drop()

要确认已删除 solarDemo，请使用 sp.listStreamProcessors() 方法列出所有可用的流处理器：

sp.listStreamProcessors()