/ /

教程：将数据从 MongoDB 连续复制到 Amazon Web Services S3

使用 Atlas Data Federation 和 Atlas 定时触发器将数据从 Atlas 集群复制到 Apache Parquet 格式的 AWS S3 存储桶。Parquet 是一种列式格式，适用于分析和机器学习工作负载，这些工作负载需要以文件而非文档的形式来处理数据。按照循环安排运行复制，以从操作集群卸载分析查询。

关于此任务

本教程使用了增量方法，这意味着每次 Trigger 运行都会复制过去 60 秒的文档。另一种方法是完全快照，即每次复制整个集合。正确的方法取决于数据量和下游消费者的要求。

本教程中的 maxFileSize 和 maxRowGroupSize 值针对测试而非生产进行了优化。对于生产工作负载，请查看 $out 阶段选项，并根据查询模式调整文件大小和分区。

开始之前

在开始本教程之前，请完成以下任务：

创建 Atlas 账户，并创建包含您要复制数据的集群。要开始，请参阅`创建集群。`
创建具有创建 IAM 角色和 S3 存储桶权限的 AWS 账户。要配置 Atlas Data Federation 所需的权限，请参阅部署联合数据库实例数据存储。
安装和配置 AWS CLI。

步骤

部署具有 S3 和 Atlas 数据存储的联合数据库实例。

联合数据库实例将多个数据源整合到单一可查询的接口中。在本教程中，您将 S3 存储桶和 Atlas 集群作为数据存储连接到同一个联合数据库实例中。连接两个数据存储可以使复制 trigger 从集群读取，并写入 S3。

部署具有 S3 数据存储的联合数据库实例。要了解具体操作，请参阅部署联合数据库实例数据存储。配置 S3 数据存储时，授予 IAM 角色 Read and write 对存储桶的访问权限，以便 Atlas Data Federation 可以写入 Parquet 文件。
将 Atlas 集群添加为联合数据库实例中的第二个数据存储。

完成这些步骤后，记下联合数据库实例服务的名称。在后续步骤中，您将需要此名称。

创建定时触发器以插入测试文档。

创建一个定时触发器，每分钟向集群插入一个新文档。这将生成测试数据，以便您验证复制 trigger 是否工作。

在 Atlas 中，前往 Triggers 页面。
1. 如果尚未显示，请从导航栏上的 Organizations 菜单中选择包含项目的组织。
2. 如果尚未显示，请从导航栏的 Projects 菜单中选择您的项目。
3. 在侧边栏中，单击 Streaming Data 标题下的 Triggers。
会显示触发器页面。
单击 Add Trigger（连接）。
选择 Scheduled 作为 Trigger Type。
在 Trigger Details 中，设立以下配置：
设置
值
Trigger Name
Create_Event_Every_Min_Trigger
Schedule Type
Basic
Interval
每 1 分钟
Event Type
Function

在 Function 部分中，选择 + New Function 并输入以下代码。将占位符值替换为 Atlas 服务、数据库和集合的名称。

exports = function () {
   const mongodb = context.services.get(
      "NAME_OF_YOUR_ATLAS_SERVICE"
   );
   const db = mongodb.db("NAME_OF_YOUR_DATABASE");
   const events = db.collection(
      "NAME_OF_YOUR_COLLECTION"
   );
   const event = events.insertOne({
      time: new Date(),
      aNumber: Math.random() * 100,
      type: "event"
   });
   return JSON.stringify(event);
};

单击 Save（连接）。
trigger 运行后，确认每分钟都有新文档出现在集群集合中。

创建一个定时触发器，用于将数据复制到 S3 。

创建一个定时触发器，运行聚合管道，使用 $out 阶段将集群中的最新文档复制到 S3 存储桶，以 Parquet 格式每分钟运行一次。

在 Triggers 页面上，单击 Add Trigger。
选择 Scheduled 作为 Trigger Type。
在 Trigger Details 中，设立以下配置：
设置
值
Trigger Name
Copy_Events_To_S3_Trigger
Schedule Type
Basic
Interval
每 1 分钟
Event Type
Function

在 Function 部分中，选择 + New Function 并输入以下代码。将占位符值替换为您的联合数据库实例服务、虚拟数据库、虚拟集合、S3 存储桶和 AWS 区域的名称。

exports = function () {
   const service = context.services.get(
      "NAME_OF_YOUR_FEDERATED_DATA_SERVICE"
   );
   const db = service.db(
      "NAME_OF_YOUR_VIRTUAL_DATABASE"
   );
   const events = db.collection(
      "NAME_OF_YOUR_VIRTUAL_COLLECTION"
   );
   const pipeline = [
      {
         $match: {
            "time": {
               $gt: new Date(
                  Date.now() - 60 * 1000
               ),
               $lt: new Date(Date.now())
            }
         }
      },
      {
         "$out": {
            "s3": {
               "bucket": "YOUR_S3_BUCKET_NAME",
               "region": "YOUR_AWS_REGION",
               "filename": "events",
               "format": {
                  "name": "parquet",
                  "maxFileSize": "10GB",
                  "maxRowGroupSize": "100MB"
               }
            }
         }
      }
   ];
   return events.aggregate(pipeline);
};

单击 Save（连接）。
trigger 运行后，确认名为 events 的 Parquet 文件出现在 S3 存储桶中。

了解详情

后退

教程：自动化集群配置

来年

禁用触发器

设置	值
Trigger Name	`Create_Event_Every_Min_Trigger`
Schedule Type	Basic
Interval	每 `1` 分钟
Event Type	Function

设置	值
Trigger Name	`Copy_Events_To_S3_Trigger`
Schedule Type	Basic
Interval	每 `1` 分钟
Event Type	Function