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

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

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

1. 部署带有 S3 数据存储的联合数据库实例。要了解具体方法，请参阅 部署联合数据库实例数据存储。配置 S3 数据存储时，授予 IAM 角色 Read and write 对存储桶的访问权限，以便 Atlas Data Federation 可以写入 Parquet 文件。

2. 将 Atlas 集群添加为联合数据库实例中的第二个数据存储。

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

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

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

1. 在 Atlas 中，前往 Triggers 页面。

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

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

   3. 在侧边栏中，单击 Streaming Data 标题下的 Triggers。

   会显示触发器页面。

2. 单击 Add Trigger（连接）。

3. 选择 Scheduled 作为 Trigger Type。

4. 在 Trigger Details 中，设立以下配置：

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

5. 在 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);
   };

6. 单击 Save（连接）。

   trigger 运行后，确认每分钟都有新文档出现在集群集合中。

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

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

1. 在 Triggers 页面上，单击 Add Trigger。

2. 选择 Scheduled 作为 Trigger Type。

3. 在 Trigger Details 中，设立以下配置：

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

4. 在 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);
   };

5. 单击 Save（连接）。

   trigger 运行后，确认命名为 events 的 Parquet 文件出现在您的 S3 存储桶中。