使用 Amazon MSK Serverless、Amazon MSK Connect 和 MongoDB Atlas 构建无服务器流式传输管道

这篇文章是与来自 MongoDB 的 Babu Srinivasan 和 Robert Walters 共同撰写的。

适用于Apache Kafka的Amazon托管流 (Amazon MSK) 是一种完全托管、高度可用的 Apache Kafka 服务。 Amazon MSK 可以轻松实时摄取和处理流数据，并在 AWS 生态系统中轻松使用这些数据。 和 亚马逊 MSK 无服务器，您可以自动配置和管理所需资源，为您的应用程序提供按需流媒体容量和存储。

Amazon MSK 还支持通过 MongoDB Atlas 等数据源集成 亚马逊 MSK 连接. MSK Connect 允许使用 Amazon MSK 将 MongoDB 数据无服务器集成 MongoDB 连接器 对于阿帕奇卡夫卡。

MongoDB Atlas 无服务器 提供可根据数据大小和吞吐量动态扩展和缩减的数据库服务，并且相应地增加和缩减成本。 它最适合需要以最少的配置进行管理的可变需求的应用程序。 它通过 MongoDB Atlas 基础架构内置的自动升级、加密、安全、指标和备份功能提供高性能和可靠性。

MSK Serverless 是 Amazon MSK 的一种集群。 就像 MongoDB Atlas Serverless 一样，MSK Serverless 会自动配置和扩展计算和存储资源。 您现在可以创建端到端的无服务器工作流。 您可以使用 MSK 无服务器和无服务器存储使用 MongoDB Atlas 构建具有无服务器摄取的无服务器流媒体管道。 此外，MSK Connect 现在支持 私有 DNS 主机名. 这允许无服务器 MSK 实例通过以下方式连接到无服务器 MongoDB 集群 AWS私有链接，为您提供平台之间的安全连接。

如果您对使用非无服务器集群感兴趣，请参阅 将 MongoDB 与 Amazon Managed Streaming for Apache Kafka (MSK) 集成.

这篇文章演示了如何使用 MSK Serverless、MSK Connect 和 MongoDB Atlas 实现无服务器流式传输管道。

解决方案概述

下图说明了我们的解决方案体系结构。

数据流开始于 亚马逊弹性计算云 (Amazon EC2) 将记录写入 MSK 主题的客户端实例。 当数据到达时，Apache Kafka 的 MongoDB 连接器实例将数据写入 MongoDB Atlas Serverless 集群中的集合。 为了两个平台之间的安全连接，在 MongoDB Atlas 集群和包含 MSK 实例的 VPC 之间创建了一个 AWS PrivateLink 连接。

这篇文章将引导您完成以下步骤：

1. 创建无服务器 MSK 集群。
2. 创建 MongoDB Atlas Serverless 集群。
3. 配置 MSK 插件。
4. 创建 EC2 客户端。
5. 配置 MSK 主题。
6. 将适用于 Apache Kafka 的 MongoDB 连接器配置为接收器。

配置无服务器 MSK 集群

要创建无服务器 MSK 集群，请完成以下步骤：

1. 在 Amazon MSK 控制台上，选择 集群 在导航窗格中。
2. 创建集群.
3. 针对 创作方法， 选择 自定义创建.
4. 针对 集群名称，输入 MongoDBMSKCluster.
5. 针对 集群类型选择 无服务器.
6. 下一页.
7. 点击 网络相关 页面，指定您的 VPC、可用区和相应的子网。
8. 记下稍后要使用的可用区和子网。
9. 下一页.
10. 创建集群.

当集群可用时，其状态变为 Active.

创建 MongoDB Atlas Serverless 集群

要创建 MongoDB Atlas 集群，请按照 开始使用阿特拉斯 教程。 请注意，出于本文的目的，您需要创建一个无服务器实例。

创建集群后，通过以下步骤配置 AWS 专用端点：

1. 点击 安保行业 菜单中选择 网络访问.
2. 点击 专用端点 标签，选择 无服务器实例.
   
3. 创建新端点.
4. 针对 无服务器实例，选择您刚刚创建的实例。
5. 确认.
6. 提供您的 VPC 端点配置并选择 下一页.
7. 创建 AWS PrivateLink 资源时，请确保指定与之前在为无服务器 MSK 实例创建网络配置时使用的完全相同的 VPC 和子网。
8. 下一页.
9. 按照说明进行操作 最终确定 页，然后选择 确认 在您的 VPC 端点创建之后。

成功后，新的专用端点将显示在列表中，如以下屏幕截图所示。

配置 MSK 插件

接下来，我们使用适用于 Apache Kafka 的 MongoDB 连接器在 Amazon MSK 中创建自定义插件。 连接器需要上传到 亚马逊简单存储服务 (Amazon S3) 存储桶，然后才能创建插件。 要下载适用于 Apache Kafka 的 MongoDB 连接器，请参阅 下载连接器 JAR 文件.

1. 在 Amazon MSK 控制台上，选择 定制插件 在导航窗格中。
2. 创建自定义插件.
3. 针对 S3 URI，输入下载的连接器的 S3 位置。
4. 创建自定义插件.

配置 EC2 客户端

接下来，让我们配置一个 EC2 实例。 我们使用这个实例来创建主题并将数据插入到主题中。 有关说明，请参阅部分 配置 EC2 客户端 在文中 将 MongoDB 与 Amazon Managed Streaming for Apache Kafka (MSK) 集成.

在 MSK 集群上创建主题

要创建 Kafka 主题，我们需要先安装 Kafka CLI。

1. 在客户端 EC2 实例上，首先安装 Java：

sudo yum install java-1.8.0

2. 接下来，运行以下命令下载 Apache Kafka：

wget https://archive.apache.org/dist/kafka/2.6.2/kafka_2.12-2.6.2.tgz

3. 使用以下命令解压缩 tar 文件：

tar -xzf kafka_2.12-2.6.2.tgz

Kafka 的发行版包括一个 bin 文件夹，其中包含可用于管理主题的工具。

4. 去 kafka_2.12-2.6.2 目录并发出以下命令以在无服务器 MSK 集群上创建 Kafka 主题：

bin/kafka-topics.sh --create --topic sandbox_sync2 --bootstrap-server <BOOTSTRAP SERVER> --command-config=bin/client.properties --partitions 2

您可以将引导服务器端点复制到 查看客户信息 无服务器 MSK 集群的页面。

您可以按照以下步骤配置 IAM 身份验证 说明.

配置接收器连接器

现在，让我们配置一个接收器连接器以将数据发送到 MongoDB Atlas Serverless 实例。

1. 在 Amazon MSK 控制台上，选择 同轴连接器 在导航窗格中。
2. 创建连接器.
3. 选择您之前创建的插件。
4. 下一页.
5. 选择您之前创建的无服务器 MSK 实例。
6. 按照以下代码输入您的连接配置：

connector.class=com.mongodb.kafka.connect.MongoSinkConnector
key.converter.schema.enable=false
value.converter.schema.enable=false
database=MongoDBMSKDemo
collection=Sink
tasks.max=1
topics=MongoDBMSKDemo.Source
connection.uri=(MongoDB Atlas Connection String Gos Here) value.converter=org.apache.kafka.connect.storage.StringConverter key.converter=org.apache.kafka.connect.storage.StringConverter

确保通过 AWS PrivateLink 连接到 MongoDB Atlas Serverless 实例。 有关详细信息，请参阅 使用 AWS PrivateLink 将应用程序安全地连接到 MongoDB Atlas 数据平面.

7. 在 访问权限 节，创建一个 AWS身份和访问管理 (IAM) 角色 所需的信任策略.
8. 下一页.
9. 指定 Amazon CloudWatch日志 作为您的日志传送选项。
10. 完成您的连接器。

当连接器状态更改为活动时，管道已准备就绪。

向 MSK 主题中插入数据

在您的 EC2 客户端上，使用 kafka-console-producer 如下：

bin/kafka-console-producer.sh --topic sandbox_sync2 --bootstrap-server <BOOTSTRAP SERVER> --producer.config=bin/client.properties

为了验证数据是否成功从 Kafka 主题流向无服务器 MongoDB 集群，我们使用 MongoDB Atlas UI。

如果遇到任何问题，请务必检查日志文件。 在此示例中，我们使用 CloudWatch 读取从 Amazon MSK 和 MongoDB Connector for Apache Kafka 生成的事件。

清理

为避免产生未来费用，请清理您创建的资源。 首先，删除 MSK 集群、连接器和 EC2 实例：

1. 在 Amazon MSK 控制台上，选择 集群 在导航窗格中。
2. 选择您的集群，然后在 行动 菜单中选择 删除.
3. 同轴连接器 在导航窗格中。
4. 选择您的连接器并选择 删除.
5. 定制插件 在导航窗格中。
6. 选择您的插件并选择 删除.
7. 在 Amazon EC2 控制台上，选择 实例 在导航窗格中。
8. 选择您创建的实例。
9. 实例状态，然后选择 终止实例.
10. 点击 亚马逊VPC 控制台，选择 端点 在导航窗格中。
11. 选择您创建的端点并在 行动 菜单中选择 删除 VPC 终端节点.

现在您可以删除 Atlas 集群和 AWS PrivateLink：

12. 登录Atlas集群控制台。
13. 导航到要删除的无服务器集群。
14. 在选项下拉菜单中，选择 终止.
15. 导航到 网络访问 部分。
16. 选择专用终结点。
17. 选择无服务器实例。
18. 在选项下拉菜单中，选择 终止.

总结

在本文中，我们向您展示了如何使用 MSK Serverless 和 MongoDB Atlas Serverless 构建无服务器流式摄取管道。 借助 MSK Serverless，您可以根据需要自动配置和管理所需资源。 我们使用部署在 MSK Connect 上的 MongoDB 连接器来无缝集成这两个服务，并使用 EC2 客户端将示例数据发送到 MSK 主题。 MSK Connect 现在支持 私有 DNS 主机名，使您能够在服务之间使用私有域名。 在本文中，连接器使用 VPC 的默认 DNS 服务器来解析可用区特定的私有 DNS 名称。 此 AWS PrivateLink 配置允许 MSK 无服务器实例和 MongoDB Atlas 无服务器实例之间的安全和私有连接。

要继续学习，请查看以下资源：

作者简介

伊戈尔·阿列克谢耶夫 是 AWS 数据和分析领域的高级合作伙伴解决方案架构师。 Igor 正在与战略合作伙伴合作，帮助他们构建复杂的 AWS 优化架构。 在加入 AWS 之前，作为一名数据/解决方案架构师，他在大数据领域实施了许多项目，包括 Hadoop 生态系统中的多个数据湖。 作为一名数据工程师，他参与了将 AI/ML 应用于欺诈检测和办公自动化的工作。

基兰马蒂 是 Amazon Web Services (AWS) 的首席产品经理，并与位于加利福尼亚州帕洛阿尔托的 Amazon Managed Streaming for Apache Kafka (Amazon MSK) 团队合作。 他热衷于构建高性能流媒体和分析服务，帮助企业实现其关键用例。

 巴布·斯里尼瓦桑（Babu Srinivasan） 是 MongoDB 的高级合作伙伴解决方案架构师。 在目前的职位上，他正在与 AWS 合作，为 AWS 和 MongoDB 解决方案构建技术集成和参考架构。 他在数据库和云技术方面拥有超过二十年的经验。 他热衷于为与跨多个地区的多个全球系统集成商 (GSI) 合作的客户提供技术解决方案。

罗伯特沃尔特斯 目前是 MongoDB 的高级产品经理。 在加入 MongoDB 之前，Rob 在 Microsoft 工作了 17 年，担任过各种职务，包括 SQL Server 团队的项目管理、咨询和技术售前。 Rob 与人合着了三项有关 SQL Server 中使用的技术的专利，并且是多本有关 SQL Server 的技术书籍的主要作者。 Rob 目前是 MongoDB 博客上的活跃博主。

• SEO 支持的内容和 PR 分发。 今天得到放大。
• 柏拉图区块链。 Web3 元宇宙智能。 知识放大。 访问这里。
• Sumber: https://aws.amazon.com/blogs/big-data/build-a-serverless-streaming-pipeline-with-amazon-msk-serverless-amazon-msk-connect-and-mongodb-atlas/