开始使用 Atlas Stream Processing

AtlasGoStream Processing在Atlas中，Go项目的 页面。

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

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

3. 在侧边栏中，单击 Services 标题下的 Stream Processing。

   此时将显示 Stream Processing 页面。

创建Atlas Stream Processing实例。

1. 点击右下角的Get Started 。 Atlas 简要说明了 Atlas Stream Processing 的核心组件。

2. 单击 Create instance 按钮。

3. 在Create a stream processing instance页面上，按如下方式配置实例：

   • Tier: SP30

   • Provider: AWS

   • Region: us-east-1

   • Instance Name: tutorialInstance

4. 单击 Create（连接）。

获取Atlas Stream Processing实例连接string 。

1. 找到Atlas Stream Processing实例的概述面板，然后单击 Connect。

2. 选择 I have the MongoDB shell installed 。

3. 从Select your mongo shell version下拉菜单中，选择最新版本的mongosh 。

4. 复制 Run your connection string in your command line 下提供的连接string 。 您将在后续步骤中用到它。

5. 单击 Close（连接）。

将 MongoDB Atlas 连接添加到连接注册表。

此连接用作我们的流数据接收器。

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

2. 在Connection Registry标签页中，单击右上角的+ Add Connection 。

3. 单击Atlas Database 。 在Connection Name字段中，输入mongodb1 。 从Atlas Cluster下拉列表中，选择一个未存储任何数据的 Atlas 集群。

4. 单击 Add connection（连接）。

验证您的流数据源是否发出消息。

您的流处理实例预配置了与名为 sample_stream_solar 的示例数据源的连接。该源可生成来自各种太阳能设备的报告流。每个报告都描述了在特定时间点观察到的单个太阳能设备的瓦数和温度，以及该设备的最大瓦数。

以下文档是一个代表性示例。

{
  device_id: 'device_8',
  group_id: 7,
  timestamp: '2024-08-12T21:41:01.788+00:00',
  max_watts: 450,
  event_type: 0,
  obs: {
    watts: 252,
    temp: 17
  },
  _ts: ISODate('2024-08-12T21:41:01.788Z'),
  _stream_meta: {
    source: {
      type: 'generated'
    }
  }
}

要验证此源是否发出消息，请以交互方式创建流处理器。

1. 打开您选择的终端应用程序。

2. 使用 连接到Atlas Stream Processing mongosh实例。

   将您在上一步中复制的 连接mongosh 粘贴到终端，其中string<atlas-stream-processing-url> URL是Atlas Stream Processing 实例的 ，<username> 是具有atlasAdmin 角色的用户。

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

   根据提示输入密码。

3. 创建流处理器。

   将以下代码复制到mongosh提示符中：

   sp.process([{"$source": {
      "connectionName": "sample_stream_solar"
   }}])

   验证来自sample_stream_solar连接的数据是否显示在控制台上，并终止该进程。

   使用sp.process()创建的流处理器在终止后不会持续存在。

创建持久流处理器。

使用聚合管道，您可以在摄取每个文档时对其进行转换。以下聚合管道以一秒的间隔得出每个太阳能设备的最高温度以及平均、中值、最大和最小瓦数。

1. 配置 $source 阶段。

   以下 $source 阶段从 sample_stream_solar 源摄入数据。

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

2. 配置 $group 阶段。

   以下 $group 阶段根据其 group_id 组织所有传入数据，累积每个 group_id 所有文档的 obs.temp 和 obs.watts 字段的值，然后派生所需的数据。

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

3. 配置 $tumblingWindow 阶段。

   为了流媒体数据执行 $group 等累加操作， Atlas Stream Processing 使用窗口来限制数据集。接下来的 $tumblingWindow 阶段将数据流分成连续的 10 秒间隔。

   这意味着，例如，当 $group 阶段计算 median_watts 的值时，它会获取在前 10 秒内摄取的具有给定 group_id 的所有文档的 obs.watts 值。

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

4. 配置 $merge 阶段。

   $merge 允许您将已处理的流数据写入 Atlas 数据库。

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

5. 创建流处理器。

   为新的流处理器指定名称，并通过按顺序列出每个阶段来声明其聚合管道。$group 阶段属于 $tumblingWindow 的嵌套管道，您不得将其包含在处理器管道定义中。

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

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

要了解有关 Atlas Stream Processing 如何写入静态数据库的更多信息，请参阅 $merge。

启动流处理器。

在mongosh中运行以下命令：

sp.solarDemo.start()

验证流处理器的输出。

要验证处理器是否处于活动状态，请在mongosh中运行以下命令：

sp.solarDemo.stats()

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

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

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

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

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

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

      会显示集群页面。

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

   显示数据浏览器。

3. 查看 MySolar 集合。

或者，您也可以使用 mongosh: 在终端中显示已处理文档的样本

sp.solarDemo.sample()

{
  _id: 10,
  max_watts: 136,
  min_watts: 130,
  avg_watts: 133,
  median_watts: 130,
  max_temp: 7,
  _stream_meta: {
    source: {
      type: 'generated'
    },
    window: {
      start: ISODate('2024-08-12T22:49:05.000Z'),
      end: ISODate('2024-08-12T22:49:10.000Z')
    }
  }
}

删除流处理器。

在mongosh中运行以下命令：

sp.solarDemo.drop()

要确认已删除avgWatts ，请列出所有可用的流处理器：

sp.listStreamProcessors()