Aws MCP Etl Pipeline

AWS MCP ETL管道项目

使用AWS模型上下文协议和Amazon Q构建智能ETL管道进行会话数据处理。

概述

该项目演示了如何使用会话式人工智能自动化复杂的ETL管道开发，将工程工作从几天减少到几个小时，同时保持安全最佳实践。

建筑

该解决方案使用AWS模型上下文协议（MCP）服务器，通过自然语言交互实现大型语言模型和AWS服务之间的无缝集成。

关键组件

• AWS MCP服务器：Redshift、S3表和数据处理
• 亚马逊Q开发者 与VS Code集成
• AWS服务：亚马逊Redshift、S3、S3表、胶水、EMR、MWAA

用例

1.数据提取到S3

数据科学家可以使用对话提示从Redshift中提取紧急数据集：

• 通过自然语言创建S3存储桶
• 查询和示例Redshift表
• 使用复杂过滤器连接表
• 使用UNLOAD命令导出数据
• 自动数据质量验证

2.迁移到S3表

数据工程师构建从Redshift到S3表的生产ETL管道：

• 使用命名空间创建S3表
• 从S3存储桶导入数据
• 验证数据加载成功
• 生成参数化的PySpark脚本

先决条件

• 具有适当IAM权限的AWS帐户
• 已配置AWS CLI v2.27+
• Visual Studio代码（1.85.0+）
• 亚马逊Q开发者扩展

安装

1.安装MCP服务器

在VS Code中安装以下MCP服务器：

• Redshift MCP服务器
• S3表MCP服务器
• 数据处理MCP服务器 （与 --allow-write 论点）

2.AWS设置

# Configure AWS CLI
aws configure

# Create Redshift Serverless workgroup (optional)
aws redshift-serverless create-workgroup \
  --workgroup-name demo-workgroup \
  --base-capacity 8

使用示例

用例1：快速数据提取

# Create S3 bucket
create new s3 standard bucket with name conversationalai-demo-

# List Redshift tables
list all tables in public schema and count the number of records in each table

# Sample data
list 10 records from orders table

# Extract filtered data
Using the Redshift UNLOAD command, extract order details where priority is '1-URGENT', '2-HIGH', or '3-MEDIUM' into CSV format in S3 bucket 'conversationalai-demo-'

# Quality check
do a quality check on the data in bucket conversationalai-demo-

用例2:S3表迁移

# Create S3 Tables
Create a new S3 table bucket called order-customer-data with namespace ns_order_customer_data

# Import data
Import all files beginning with 'order_customer_data' into namespace 'ns_order_customer_data' from S3 bucket 'conversationalai-demo-'

# Validate import
check if import completed

# Generate production script
Create a PySpark script that takes order date as parameter and imports data from S3 to S3 Tables

生成的脚本

该项目包括通过对话式AI生成的示例脚本：

• pyspark_etl_script.py -用于生产ETL的参数化PySpark脚本
• data_quality_check.py -自动数据验证
• setup_infrastructure.sh -基础设施设置脚本

最佳实践

1.快速工程

• 在AI提示中要准确明确
• 结构提示，上下文清晰
• 将复杂的需求分解为可管理的块
• 测试并迭代以获得最佳结果

2.安全第一

• 遵循IAM角色的最小特权原则
• 授予最低限度的必要权限
• 定期审计访问控制
• 永远不要为了方便而牺牲安全性

3.数据验证

• 始终在生产前验证AI生成的代码
• 对转换进行彻底测试
• 验证数据的一致性和准确性
• 仔细检查生成的SQL查询

项目结构

aws-mcp-etl-pipeline/
├── README.md
├── docs/
│   ├── architecture.md
│   ├── setup-guide.md
│   └── troubleshooting.md
├── scripts/
│   ├── pyspark_etl_script.py
│   ├── data_quality_check.py
│   └── setup_infrastructure.sh
├── examples/
│   ├── use-case-1-prompts.md
│   └── use-case-2-prompts.md
└── iam/
    └── required-permissions.json

益处

• 时间缩短：编码工作天数减少到小时
• 自然语言接口：通过对话提示进行复杂查询
• 自动代码生成：AI创建优化的SQL和PySpark脚本
• 质量保证：内置数据验证和质量检查
• 生产就绪：生成用于生产的参数化脚本

清理

为避免收费，请在测试后清理资源：

# Delete S3 buckets
aws s3 rb s3://conversationalai-demo- --force

# Delete S3 Tables
# (Use conversational AI or AWS CLI)

# Delete Redshift Serverless workgroup
aws redshift-serverless delete-workgroup --workgroup-name demo-workgroup

贡献

1. 分叉存储库
2. 创建要素分支
3. 进行更改
4. 添加测试和文档
5. 提交拉取请求

许可证

此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。

致谢

基于Avijit Goswami和Ajit Tandale的AWS博客文章：“使用AWS模型上下文协议和Amazon Q构建智能ETL管道”。

支持

如有疑问和支持：

• 在此存储库中打开问题
• 检查 故障排除指南
• 查看AWS MCP文档