学习目标
通过本章学习，你将能够：
- 综合运用DSPy技术：集成前面章节学到的所有DSPy核心概念和技术
- 构建完整系统架构：设计和实现一个生产级的智能问答系统
- 实现端到端流程：从数据处理到模型训练，从API设计到前端展示
- 掌握项目最佳实践：代码组织、错误处理、性能优化、监控告警
- 学会系统部署和维护：容器化部署、自动化测试、持续集成
2025/9/14大约 24 分钟
2025/9/1小于 1 分钟
第 1 章：CDK 概述与环境准备
学习目标：
- 理解什么是 AWS CDK 以及它解决的问题
- 了解 CDK 与 CloudFormation 的关系
- 掌握 CDK 开发环境的搭建
- 完成第一个 CDK 应用的创建
简要说明：介绍 CDK 的核心概念，对比传统 IaC 工具的优劣势，建立开发环境并创建 Hello World 应用。
第 2 章：CDK 核心概念
2025/9/1大约 4 分钟
学习目标
- 理解什么是 AWS CDK 以及它解决的问题
- 了解 CDK 与 CloudFormation 的关系
- 掌握 CDK 开发环境的搭建
- 完成第一个 CDK 应用的创建
知识点总结
什么是 AWS CDK
AWS Cloud Development Kit (CDK) 是一个开源软件开发框架，用于使用熟悉的编程语言来定义云基础设施资源。CDK 支持 TypeScript、JavaScript、Python、Java、C#/.Net 和 Go。
2025/9/1大约 4 分钟
学习目标
- 掌握 App、Stack、Construct 三大核心概念
- 理解 CDK 的层级结构和组织方式
- 学会使用 CDK CLI 基本命令
- 了解 CDK 的生命周期和部署流程
核心概念详解
CDK 三大核心概念
1. App（应用）
App 是 CDK 应用的根容器，代表整个 CDK 应用程序。
2025/9/1大约 6 分钟
学习目标
- 掌握 L1、L2、L3 Constructs 的区别和使用场景
- 学会创建和配置基础 AWS 资源（S3、Lambda、IAM）
- 理解资源间的依赖关系
- 掌握 CDK 中的参数传递和配置管理
Constructs 层级深入理解
三层架构对比
选择合适的 Construct 层级
| 场景 | 推荐层级 | 原因 |
|---|---|---|
| 快速原型开发 | L3 | 减少代码量，快速验证 |
| 生产环境标准配置 | L2 | 平衡易用性和控制力 |
| 特殊配置需求 | L1 | 完全控制所有属性 |
| 遗留系统迁移 | L1 | 精确映射现有配置 |
2025/9/1大约 8 分钟
学习目标
- 掌握自定义 Construct 的创建方法
- 学会使用 CDK Patterns 和最佳实践
- 理解 Cross-Stack References 的使用
- 掌握 CDK Context 和 Feature Flags 的应用
自定义 Construct 开发
Construct 设计原则
创建自定义 Construct
2025/9/1大约 9 分钟
学习目标
- 掌握 VPC 网络架构的 CDK 实现
- 学会配置 EC2、RDS、ElastiCache 等计算和存储服务
- 理解 API Gateway、Lambda 的无服务器架构实现
- 掌握 CloudWatch、SNS、SQS 等监控和消息服务配置
VPC 网络架构
完整的 VPC 设置
from aws_cdk import (
Stack,
aws_ec2 as ec2,
aws_logs as logs,
CfnOutput
)
from constructs import Construct
class NetworkingStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# 创建 VPC
self.vpc = ec2.Vpc(
self,
"MainVPC",
vpc_name="main-vpc",
max_azs=3, # 使用 3 个可用区
cidr="10.0.0.0/16",
# 定义子网配置
subnet_configuration=[
# 公有子网 - 用于负载均衡器、NAT 网关
ec2.SubnetConfiguration(
name="Public",
subnet_type=ec2.SubnetType.PUBLIC,
cidr_mask=24 # 10.0.0.0/24, 10.0.1.0/24, 10.0.2.0/24
),
# 私有子网 - 用于应用服务器
ec2.SubnetConfiguration(
name="Private",
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS,
cidr_mask=24 # 10.0.10.0/24, 10.0.11.0/24, 10.0.12.0/24
),
# 数据库子网 - 完全隔离
ec2.SubnetConfiguration(
name="Database",
subnet_type=ec2.SubnetType.PRIVATE_ISOLATED,
cidr_mask=24 # 10.0.20.0/24, 10.0.21.0/24, 10.0.22.0/24
)
],
# 启用 DNS
enable_dns_hostnames=True,
enable_dns_support=True,
# NAT 网关配置
nat_gateways=1, # 生产环境建议每个 AZ 一个
# 流日志
flow_logs={
"FlowLogsCloudWatch": ec2.FlowLogOptions(
destination=ec2.FlowLogDestination.to_cloud_watch_logs(
log_group=logs.LogGroup(
self,
"VpcFlowLogsGroup",
retention=logs.RetentionDays.ONE_MONTH
)
),
traffic_type=ec2.FlowLogTrafficType.ALL
)
}
)
# 创建安全组
self._create_security_groups()
# 创建 VPC 端点（减少 NAT 网关费用）
self._create_vpc_endpoints()
# 输出网络信息
self._create_outputs()
def _create_security_groups(self):
"""创建安全组"""
# Web 层安全组
self.web_sg = ec2.SecurityGroup(
self,
"WebSecurityGroup",
vpc=self.vpc,
description="Security group for web servers",
allow_all_outbound=True
)
# 允许 HTTP/HTTPS 流量
self.web_sg.add_ingress_rule(
peer=ec2.Peer.any_ipv4(),
connection=ec2.Port.tcp(80),
description="Allow HTTP"
)
self.web_sg.add_ingress_rule(
peer=ec2.Peer.any_ipv4(),
connection=ec2.Port.tcp(443),
description="Allow HTTPS"
)
# 应用层安全组
self.app_sg = ec2.SecurityGroup(
self,
"AppSecurityGroup",
vpc=self.vpc,
description="Security group for application servers",
allow_all_outbound=True
)
# 只允许来自 Web 层的流量
self.app_sg.add_ingress_rule(
peer=self.web_sg,
connection=ec2.Port.tcp(8080),
description="Allow traffic from web tier"
)
# 数据库层安全组
self.db_sg = ec2.SecurityGroup(
self,
"DatabaseSecurityGroup",
vpc=self.vpc,
description="Security group for database servers"
)
# 只允许来自应用层的数据库连接
self.db_sg.add_ingress_rule(
peer=self.app_sg,
connection=ec2.Port.tcp(5432),
description="Allow PostgreSQL from app tier"
)
self.db_sg.add_ingress_rule(
peer=self.app_sg,
connection=ec2.Port.tcp(3306),
description="Allow MySQL from app tier"
)
# 缓存层安全组
self.cache_sg = ec2.SecurityGroup(
self,
"CacheSecurityGroup",
vpc=self.vpc,
description="Security group for cache servers"
)
self.cache_sg.add_ingress_rule(
peer=self.app_sg,
connection=ec2.Port.tcp(6379),
description="Allow Redis from app tier"
)
# 管理访问安全组
self.bastion_sg = ec2.SecurityGroup(
self,
"BastionSecurityGroup",
vpc=self.vpc,
description="Security group for bastion host"
)
# 只允许来自公司 IP 的 SSH 访问
self.bastion_sg.add_ingress_rule(
peer=ec2.Peer.ipv4("203.0.113.0/24"), # 替换为实际的公司 IP 范围
connection=ec2.Port.tcp(22),
description="Allow SSH from company network"
)
def _create_vpc_endpoints(self):
"""创建 VPC 端点以减少 NAT 网关使用"""
# S3 网关端点
self.vpc.add_gateway_endpoint(
"S3Endpoint",
service=ec2.GatewayVpcEndpointAwsService.S3,
subnets=[
ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_ISOLATED)
]
)
# DynamoDB 网关端点
self.vpc.add_gateway_endpoint(
"DynamoDbEndpoint",
service=ec2.GatewayVpcEndpointAwsService.DYNAMODB,
subnets=[
ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS),
ec2.SubnetSelection(subnet_type=ec2.SubnetType.PRIVATE_ISOLATED)
]
)
# 接口端点安全组
endpoint_sg = ec2.SecurityGroup(
self,
"VpcEndpointSG",
vpc=self.vpc,
description="Security group for VPC endpoints"
)
endpoint_sg.add_ingress_rule(
peer=ec2.Peer.ipv4(self.vpc.vpc_cidr_block),
connection=ec2.Port.tcp(443),
description="Allow HTTPS from VPC"
)
# 常用服务的接口端点
services = [
ec2.InterfaceVpcEndpointAwsService.ECR,
ec2.InterfaceVpcEndpointAwsService.ECR_DOCKER,
ec2.InterfaceVpcEndpointAwsService.LAMBDA,
ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER,
ec2.InterfaceVpcEndpointAwsService.SSM,
ec2.InterfaceVpcEndpointAwsService.SSM_MESSAGES,
ec2.InterfaceVpcEndpointAwsService.EC2_MESSAGES
]
for service in services:
self.vpc.add_interface_endpoint(
f"{service.name}Endpoint",
service=service,
subnets=ec2.SubnetSelection(
subnet_type=ec2.SubnetType.PRIVATE_WITH_EGRESS
),
security_groups=[endpoint_sg]
)
def _create_outputs(self):
"""创建输出"""
CfnOutput(self, "VpcId", value=self.vpc.vpc_id)
CfnOutput(self, "VpcCidr", value=self.vpc.vpc_cidr_block)
CfnOutput(
self, "PublicSubnetIds",
value=",".join([s.subnet_id for s in self.vpc.public_subnets])
)
CfnOutput(
self, "PrivateSubnetIds",
value=",".join([s.subnet_id for s in self.vpc.private_subnets])
)
CfnOutput(
self, "DatabaseSubnetIds",
value=",".join([s.subnet_id for s in self.vpc.isolated_subnets])
)2025/9/1大约 10 分钟
学习目标
- 掌握 CDK 应用的单元测试编写方法
- 理解快照测试和细粒度测试的区别和用途
- 学会使用 CDK 内置的测试工具和断言
- 掌握 CDK 应用的调试技巧和故障排除方法
- 了解集成测试和端到端测试策略
知识点总结
CDK 测试概述
CDK 支持多种层次的测试，从单元测试到集成测试，确保基础设施代码的质量和可靠性。
2025/9/1大约 8 分钟
学习目标
- 掌握 CDK 多环境部署策略和最佳实践
- 理解 CDK 应用的生命周期管理
- 学会使用 CDK Pipelines 实现 CI/CD
- 掌握配置管理和参数化部署
- 了解蓝绿部署、金丝雀发布等高级部署策略
知识点总结
环境管理概述
在企业级应用中，通常需要管理多个环境：开发(dev)、测试(test)、预发布(staging)、生产(prod)。CDK 提供了灵活的机制来管理这些不同环境的差异。
2025/9/1大约 12 分钟