用 AWS CDK 构建 CodeCommit 到 AWS Glue 的自动化 ETL 部署流水线—

【免费下载链接】aws-cdk-examplesExample projects using the AWS CDK项目地址https://gitcode.com/gh_mirrors/aw/aws-cdk-examples点击查看免费下载导读本文以 aws-cdk-examples 仓库中 typescript/codepipeline-glue-deploy 示例为对象深入讲解如何用 AWS CDKTypeScript把 AWS CodeCommit、AWS CodePipeline、AWS Lambda 与 AWS Glue 组合成一条代码推送即触发 ETL 作业的自动化流水线。读完本文你将掌握该示例的端到端工作流程、CDK 栈中各资源的创建方式与参数含义、Lambda 处理器从 CodeCommit 取文件到启动 Glue 作业的完整调用链以及每一步背后的 IAM 权限模型可直接以此为基线定制自己的数据管道。一、模式概述与适用场景该示例演示了一个典型的CI/CD 驱动数据管道模式当开发者把 ETL抽取-转换-加载代码提交并推送到远程 AWS CodeCommit 仓库后会立即触发一条 CodePipeline 流水线流水线中的 Lambda 动作负责用最新代码启动一个 AWS Glue 作业由 Glue 执行 ETL 任务。正如 README 所说这个方案非常适合业务方、开发者和数据工程师希望在变更一提交并推送到目标仓库后立刻启动作业的场景它显著提升了自动化程度与可复现性避免了手工启动作业、作业生命周期管理过程中容易出现的错误README.md。二、端到端工作流程五步闭环示例 README 给出了清晰的处理链路完整流程如下开发者或数据工程师修改 ETL 代码提交并推送到 AWS CodeCommit 仓库推送动作触发 CodePipeline 流水线流水线调用 Lambda 函数Lambda 调用codecommit:GetFile从仓库读取文件并把文件上传到 Amazon S3Lambda 函数使用这段 ETL 代码启动一个新的 AWS Glue 作业Lambda 函数向流水线回报成功结果流水线结束。下图是示例仓库自带的架构示意图image.png直观展示了从 CodeCommit 推送到 Glue 作业启动的完整链路需要说明的是这里的推送触发由 CodeCommit 源动作Source Action与 CodePipeline 的事件监听机制共同完成属于仓库内置的 CI/CD 集成无需额外自建触发器。三、项目结构与关键文件示例目录结构如下typescript/codepipeline-glue-deploy/ ├── bin/ │ └── app.ts # CDK 应用入口 ├── lib/ │ └── codepipeline-glue-deploy-stack.ts # 核心基础设施栈 ├── lambda_etl_launch/ │ └── lambda_etl_launch.py # 流水线中执行取代码→建作业→启作业的 Lambda 处理器 ├── etl/ │ └── etl.py # 作为初始 ETL 代码会被推送到 CodeCommit 仓库 ├── cdk.json # CDK Toolkit 配置 ├── package.json / tsconfig.json # 依赖与 TypeScript 编译配置 └── jest.config.js # Jest 单测配置入口文件 bin/app.ts 中通过new CodepipelineGlueDeployStack(app, CodepipelineGlueDeploy, {})实例化栈而 cdk.json 的app字段指定了应用执行方式npx tsc npx tsx bin/app.ts即先编译 TypeScript 再运行入口。依赖方面示例使用aws-cdk-lib^2.262.1、constructs^10.0.0与aws-cdkCLI见 package.json。四、环境准备与常用命令README 提供了完整的本地运行命令清单按顺序执行即可把示例部署到默认 AWS 账户/区域命令作用aws configure配置 AWS 账户访问凭证npm run watch监听源码变化并增量编译对应tsc -wnpm run test执行 Jest 单元测试cdk deploy --parameters glueJobGlue Job Name部署栈到默认 AWS 账户/区域并传入 Glue 作业名参数cdk diff对比已部署栈与当前状态的差异cdk synth生成合成的 CloudFormation 模板其中--parameters glueJob...对应栈内定义的 CloudFormation 参数glueJob见下文第五小节部署时必须显式提供 Glue 作业名项目内也内置了 jest.config.js 与 jest 开发依赖供快照/单元测试使用。五、CDK 基础设施栈深度解析核心代码集中在 lib/codepipeline-glue-deploy-stack.ts。该栈从下到上依次创建了参数、仓库、加密、存储、角色、函数与流水线每一块都值得逐一拆解。5.1 Glue 作业名CloudFormation 参数const glueJob new CfnParameter(this, glueJob, { type: String, description: The name of the Glue job, });示例用CfnParameter声明了一个字符串参数glueJob这就是部署命令中--parameters glueJob...的由来。它的值会在流水线 Deploy 阶段通过userParameters传给 Lambda用于拼接最终 Glue 作业名。5.2 CodeCommit 仓库与初始 ETL 代码const etlRepository new codecommit.Repository(this, EtlRepository, { repositoryName: EtlRepository, code: codecommit.Code.fromDirectory(etl/), description: EtlRepository });栈会创建名为EtlRepository的 CodeCommit 仓库并直接使用 CDK 的Code.fromDirectory(etl/)把本地 etl/etl.py 作为初始内容导入仓库。该文件是一个最小可运行的 Glue 脚本先from awsglue.context import GlueContext、from pyspark.context import SparkContext再创建GlueContext并打印作为后续 ETL 逻辑的占位骨架。5.3 制品存储桶KMS 加密访问日志强制 SSL流水线制品Artifact需要落盘示例对存储做了完整的加固const pipelineArtifactStoreEncryptionKey new Key(this, pipelineArtifactStoreEncryptionKey, { removalPolicy: RemovalPolicy.DESTROY, enableKeyRotation: true }); const pipelineArtifactStoreBucket new Bucket(this, XXXXXXXXXXXXXXXXXXXXXXXXXXX, { removalPolicy: RemovalPolicy.DESTROY, autoDeleteObjects: true, encryption: BucketEncryption.KMS, encryptionKey: pipelineArtifactStoreEncryptionKey, serverAccessLogsPrefix: access-logs, enforceSSL: true });要点如下独立创建 KMS 密钥并开启enableKeyRotation: true自动轮换S3 桶使用BucketEncryption.KMS服务端加密并绑定该密钥serverAccessLogsPrefix: access-logs开启服务端访问日志前缀enforceSSL: true强制 HTTPS 访问autoDeleteObjects: true配合RemovalPolicy.DESTROY便于示例销毁栈时自动清空对象注意仅适用于非生产测试环境。5.4 Glue 执行角色const glueRole new Role(this, GlueRole, { assumedBy: new ServicePrincipal(glue.amazonaws.com), }); glueRole.addToPrincipalPolicy( new PolicyStatement({ actions: [glue:CreateJob, glue:StartJobRun], effect: Effect.ALLOW, resources: [*] }) );角色由glue.amazonaws.com服务主体担任被授予glue:CreateJob与glue:StartJobRun权限同时栈还通过pipelineArtifactStoreEncryptionKey.grantEncryptDecrypt(glueRole)与pipelineArtifactStoreBucket.grantReadWrite(glueRole)让该角色能够加解密制品密钥、读写制品桶从而支撑 Glue 从 S3 读取脚本。5.5 Lambda 函数与环境变量const lambda new Function(this, lambda, { code: Code.fromAsset(lambda_etl_launch), handler: lambda_etl_launch.lambda_handler, runtime: Runtime.PYTHON_3_12, environment: { REPOSITORY_NAME: etlRepository.repositoryName, FILENAME: etl.py } });Lambda 使用 Python 3.12 运行时代码来自 lambda_etl_launch 目录入口为lambda_etl_launch.lambda_handler。两个环境变量REPOSITORY_NAME目标 CodeCommit 仓库名与FILENAME要读取的 ETL 文件名etl.py把仓库/文件名这类运行期信息从代码中解耦出来。5.6 流水线Source Deploy 两阶段编排const pipelineArtifactStore new Artifact(); const pipeline new Pipeline(this, Pipeline, { pipelineName: pipeline, artifactBucket: pipelineArtifactStoreBucket, crossAccountKeys: true, enableKeyRotation: true, pipelineType: PipelineType.V2, stages: [ { stageName: Source, actions: [ new CodeCommitSourceAction({ actionName: Source, repository: etlRepository, branch: main, output: pipelineArtifactStore, }) ]}, { stageName: Deploy, actions: [ new LambdaInvokeAction({ actionName: Deploy, lambda: lambda, inputs: [pipelineArtifactStore], userParameters: { glue_job_name: glueJob.valueAsString, glue_role: glueRole.roleName } }) ]}, ] });流水线类型为PipelineType.V2并开启crossAccountKeys与enableKeyRotationSource 阶段CodeCommitSourceAction监听EtlRepository的main分支推送即触发输出制品交给下一阶段Deploy 阶段LambdaInvokeAction调用上述 Lambda并把glue_job_name来自glueJob参数与glue_roleGlue 角色名以userParametersJSON 形式传给 Lambda 处理器流水线角色被授予etlRepository.grantPull拉取仓库、lambda.grantInvoke调用函数以及pipelineArtifactStoreEncryptionKey.grantEncryptDecrypt加解密制品密钥三项权限。六、Lambda 处理器从 CodeCommit 到 Glue 的调用链README 中第 3、4 步描述的行为在 lambda_etl_launch/lambda_etl_launch.py 中有完整实现。处理器被 CodePipeline 以作业Job方式调用事件结构为event[CodePipeline.job]处理逻辑可拆成六个环节解析作业数据从job[data][actionConfiguration][configuration][UserParameters]读取glue_job_name、glue_role等用户参数JSON 字符串需json.loads反序列化定位制品从data[inputArtifacts][0]拿到 S3 位置bucketName、objectKey并用source_code_artifact[revision]作为 commit id文件名由环境变量FILENAME提供读取 CodeCommit 文件调用codecommit.get_file(repositoryName..., commitSpecifiercommit_id, filePathfilename)获取 ETL 代码内容上传到 S3调用s3.put_object把fileContent写入制品桶并校验ResponseMetadata[HTTPStatusCode] 200失败即抛异常创建并启动 Glue 作业s3_script_location fs3://{artifact_bucket}/{file_key} glue_job_name_id artifact_key.split(/)[-1:][0] glue_job_name f{user_params[glue_job_name]}_{glue_job_name_id} default_arguments {} if additional_python_modules in user_params: default_arguments[--additional-python-modules] user_params[additional_python_modules] create_job_resp glue.create_job( Nameglue_job_name, Roleuser_params[glue_role], Command{Name: glueetl, ScriptLocation: s3_script_location}, DefaultArgumentsdefault_arguments, GlueVersion4.0 ) glue.start_job_run(JobNamecreate_job_resp[Name], Arguments{})作业名的最终形态是参数值下划线制品 key 最后一段避免同名冲突GlueVersion4.0指定 Glue 4.0 运行时命令类型为glueetl若userParameters中带additional_python_modules会映射为--additional-python-modules参数传入DefaultArguments便于为作业附加 Python 模块。回报执行结果成功后调用pipeline.put_job_success_result(jobIdjob[id])任何异常则进入except分支调用pipeline.put_job_failure_result携带failureDetailstype: JobFailed、错误消息与context.aws_request_id作为外部执行 ID从而让流水线正确感知作业成功或失败。七、权限模型剖析整个示例的 IAM 关系在栈源码中非常清晰可归纳为三组Glue 角色glue.amazonaws.comglue:CreateJob、glue:StartJobRun资源为*并对制品 KMS 密钥grantEncryptDecrypt、对制品 S3 桶grantReadWrite以便创建作业、从 S3 读取脚本Lambda 角色被显式授予iam:PassRole资源限定为glueRole.roleArn即只能把 Glue 角色传递给作业、制品桶grantReadWrite与制品密钥grantEncryptDecrypt流水线角色etlRepository.grantPull、lambda.grantInvoke、制品密钥grantEncryptDecrypt。从源码结构看Lambda 处理器还调用了codecommit:get_file而本栈为 Lambda 角色声明的策略集中在iam:PassRole、S3 读写与 KMS 加解密上并未显式追加 CodeCommit 读取权限若在严格策略环境下部署可按需为 Lambda 执行角色补充codecommit:GetFile等读取类权限CodeCommit 仓库的拉取权限grantPull在栈中授予的是流水线角色两者作用对象不同。这一细节可作为自行扩展示例时的关注点。八、自动化、可扩展性与定制方向正如 README Automation and scale 一节所述该示例提供的是可定制、可扩展的基线实现。基于当前仓库源码以下方向都具备清晰的改造入口替换或扩充 ETL 代码直接修改 etl/etl.py仓库初始化时会将新内容导入 CodeCommit扩展 Lambda 处理逻辑在 lambda_etl_launch.py 中增加参数透传如additional_python_modules、作业参数Arguments、资源标签等调整流水线阶段在 codepipeline-glue-deploy-stack.ts 的stages数组中加入构建、测试或审批阶段或把CodeCommitSourceAction换成 GitHub、S3 等其他源动作加固安全基线Glue 角色当前对glue:CreateJob/glue:StartJobRun使用resources: [*]生产环境可收敛为指定作业 ARN 范围。参考文档官方主题索引AWS Glue添加作业Adding jobs in AWS GlueCodePipeline在流水线中调用 AWS Lambda 函数Invoke an AWS Lambda function in CodePipelineCodePipeline源操作集成Source action integrations in CodePipeline以上主题的详细说明可参见示例 README.md 的 References 一节部署时可结合cdk diff/cdk synth先检查生成的 CloudFormation 模板再执行cdk deploy --parameters glueJobGlue Job Name完成上线。赞分享【免费下载链接】aws-cdk-examplesExample projects using the AWS CDK项目地址https://gitcode.com/gh_mirrors/aw/aws-cdk-examples点击查看免费下载相关推荐aws-cdk-examples .NET Examples 实战指南用 C 编写、构建并部署 AWS CDK 应用aws cdk examples .NET Examples 实战指南用 C 编写、构建并部署 AWS CDK 应用本文基于 csharp/README.m使用 AWS CDK 构建 CodePipeline 流水线CodeCommit 到 ECS Fargate 的 Docker 镜像构建与蓝绿部署实战使用 AWS CDK 构建 CodePipeline 流水线CodeCommit 到 ECS Fargate 的 Docker 镜像构建与蓝绿部署实战本篇技用 AWS CDK 部署 AWS Transfer Family SFTP 服务器aws-cdk-examples 实战指南用 AWS CDK 部署 AWS Transfer Family SFTP 服务器aws cdk examples 实战指南导读本文以 aws cdk e上一篇Ideogram-4ComfyUI专用模型包深度解析开启AI绘图新体验下一篇Bilibili-Evolved终极响应式设计指南从设计理念到代码实现创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

用 AWS CDK 构建 CodeCommit 到 AWS Glue 的自动化 ETL 部署流水线——aws-cdk-examples 实战解析

相关推荐

SensorsCalibration 实战：Ceres 通用无约束最小化框架 GradientProblem 与 GradientProblemSolver 完全指南

不止替换SQL关键词：KFS如何拆解并重构异构数据库DDL

Rust 全局声明体系重构：RFC 0246 对 `const` 与 `static` 的划分及其在现代 Rust 中的演进

更多资讯

PoB离线Build规划：天赋树与伤害计算一次算清

微信API发送失败别盲目重试：个人微信二次开发判断法

CKEditor 4 聚焦状态下销毁内联编辑器的空引用崩溃：`isInline of null` 问题与 Bender 手动测试解析

如何在电脑上运行《索尼克：释放》：Unleashed Recompiled 静态重编译版完整上手指南

叉积判向：从三角形朝向到背面剔除的工程实战

Wing 语言 inflight Handler 单例语义剖析：以 `inflight_handler_singleton.test.w` 快照测试为例

今日更新

防关联浏览器哪个好？三款产品实测差异解析

2026钢铁行业AI转型系统选型对比：多基地排产、大宗原燃料采购与质量追溯，六家厂商能力差异在哪里

影刀RPA深度实战：大模型Fine-tuning定制RPA专用Prompt与指令生成

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？