使用 AWS Step Functions 创建和运行 AWS DMS 任务 数据库博客

• 2026-01-27 12:57:09
• 项目展示

使用 AWS Step Functions 创建并运行 AWS DMS 任务

关键要点

AWS DMSAWS 数据库迁移服务是一个托管服务，能够帮助用户轻松、安全地将数据库迁移到 AWS。它支持多种数据库源与目标，包括 Amazon RDS 、 Amazon Aurora 、 Amazon Redshift 、 Amazon S3 等。通过 AWS DMS，您可以将数据迁移到大多数流行的商业和开源数据库中。

AWS Step Functions 是一个无服务器工作流编排服务，使您能够协调和可视化多个 AWS 服务，形成无服务器工作流。它提供了一个图形界面，以 JSON 为基础的语言称为 Amazon States Language来定义工作流的流程和条件。

在本文中，我们将探讨如何使用 Step Functions 创建和编排 AWS DMS 任务。

解决方案概览

在本文中，我们将 S3 桶作为数据源，将 Amazon Aurora PostgreSQLCompatible Edition 作为目标数据库实例。通过将 AWS Lambda 与 Step Functions 集成，可以自动化 AWS DMS 任务的创建。这种解决方案提供了一个端到端的管道，以自动化的方式迁移数据。使用 Step Functions 创建 AWS DMS 任务，有助于简化过程，并获得以下优势：

以下图表展示了这一架构。

工作流包括以下步骤：

前提条件

要开始操作，您需要完成以下前提条件：

接下来的部分将为您提供自动化 AWS DMS 任务的逐步指南。

创建源和目标 AWS DMS 端点

我们使用 AWS DMS 端点 来定义迁移任务中的源和目标数据库。端点包含连接数据库所需的信息，如服务器名称、凭据、端口和数据库名。您可以使用 AWS DMS 控制台或 AWS CLI 创建源和目标端点。

当 使用 Amazon S3 作为源 时，源数据文件必须以 CSV 格式存储。例如，/schemaName/ltfoldernamegt/tableName/LOAD001csv。比如，您可以将数据文件存放在 S3 路径 s3//ltbucketNamegt/dmsfolder/dbo/ratings/ratingscsv。以下截图显示了 ratingscsv 文件中的数据示例。

以下截图展示了在 AWS DMS 控制台上创建目标端点的示例设置。当 使用 PostgreSQL 数据库作为目标 时，请提供必要的详细信息，例如 PostgreSQL 服务器端点、数据库凭据和其他连接设置。

创建一个 IAM 角色以调用 Step Functions

创建一个 IAM 角色，其权限允许从 Lambda 函数调用 Step Functions 状态机。本文中，状态机名称为 DMSS3postgresstepFunctions。Lambda 执行角色应具备与 Step Functions 交互的权限，包括必要的 startexecution 操作。以下是示例代码：

json{ Version 20121017 Statement [ { Effect Allow Action [ statesStartExecution ] Resource [ arnawsstatesuseast1stateMachineDMSS3postgresstepFunctions ] } ]}

创建 Step Functions 状态机

要为创建 AWS DMS 复制任务创建 Step Functions 工作流，您可以通过 Step Functions 控制台或 AWS CLI 创建名为 DMSS3postgresstepFunctions 的状态机。

使用创建 AWS DMS 任务所需的输入参数，如源和目标数据库端点、复制实例和选择规则。这些参数作为输入传递给状态机工作流。

以下截图展示了在控制台中使用空模板输入状态机代码的示例。

本文使用以下状态机代码：

json{ Comment A description of my state machine StartAt CreateReplicationTask States { CreateReplicationTask { Type Task Next Wait for Create Parameters { MigrationType fullload ReplicationInstanceArn arnawsdmsuseast102repBU547HLPVA7TGKWOJACENAUPKM ReplicationTaskIdentifier Myblog SourceEndpointArn arnawsdmsuseast102endpointRLQI6XJUJW56TSJVEWRQOK6R6BAGO45BIDEFG2Y TableMappings { rules [ { ruletype selection ruleid 1 rulename 1 objectlocator { schemaname dbo tablename } ruleaction include } ] } TargetEndpointArn arnawsdmsuseast102endpointUM2W53TDKBTP5U742KDUADE4LEULCCP5VC6ZVLY } Resource arnawsstatesawssdkdatabasemigrationcreateReplicationTask } Wait for Create { Type Wait Seconds 70 Next DescribeReplicationTask } DescribeReplicationTask { Type Task Next Get Task Status Parameters { Filters [ { Name replicationtaskarn Values StatesArray(ReplicationTaskReplicationTaskArn) } ] Marker MaxRecords 20 WithoutSettings false } Resource arnawsstatesawssdkdatabasemigrationdescribeReplicationTasks ResultSelector { Status ReplicationTasks[0]Status ReplicationTaskArn ReplicationTasks[0]ReplicationTaskArn } } Get Task Status { Type Choice Choices [ { Or [ { Variable Status StringEquals stopped } { Variable Status StringEquals ready } ] Next StartReplicationTask } { Variable Status StringEquals failed Next SNS Publish Failed } ] Default StartReplicationTask } StartReplicationTask { Type Task Parameters { ReplicationTaskArn ReplicationTaskArn StartReplicationTaskType reloadtarget } Resource arnawsstatesawssdkdatabasemigrationstartReplicationTask Next Wait for Start ResultSelector { Status ReplicationTaskStatus ReplicationTaskArn ReplicationTaskReplicationTaskArn } } Wait for Start { Type Wait Seconds 10 Next Get Task load Status } Get Task load Status { Type Choice Choices [ { Or [ { Variable Status StringMatches starting } { Variable Status StringMatches running } ] Next DescribeReloadStatus } { Variable Status StringMatches failed Next SNS Publish Failed } { Or [ { Variable Status StringMatches stopped } { Variable Status StringMatches ready } ] Next SNS Publish } ] Default SNS Publish } DescribeReloadStatus { Type Task Next Wait for Start Parameters { Filters [ { Name replicationtaskarn Values StatesArray(ReplicationTaskArn) } ] Marker MaxRecords 20 WithoutSettings false } Resource arnawsstatesawssdkdatabasemigrationdescribeReplicationTasks ResultSelector { Status ReplicationTasks[0]Status ReplicationTaskArn ReplicationTasks[0]ReplicationTaskArn } } SNS Publish { Type Task Resource arnawsstatessnspublish Parameters { TopicArn arnawssnsuseast102CaseTestSNS Message } Next DeleteReplicationTask ResultPath null } SNS Publish Failed { Type Task Resource arnawsstatessnspublish Parameters { TopicArn arnawssnsuseast102CaseTestSNS Message } ResultPath null End true } DeleteReplicationTask { Type Task End true Parameters { ReplicationTaskArn ReplicationTaskArn } Resource arnawsstatesawssdkdatabasemigrationdeleteReplicationTask } } TimeoutSeconds 300}

创建一个用于访问 S3 桶的 Lambda 函数的 IAM 角色

创建一个 IAM 角色，授予 Lambda 函数访问和操作 S3 桶的权限。该角色通常在创建由 S3 事件触发或需要对 S3 对象执行操作的 Lambda 函数时使用。有关更多信息，请参见 如何允许我的 Lambda 函数访问我的 Amazon S3 桶？。

创建 Lambda 函数

要从 Lambda 函数中调用 Step Functions，可以使用 AWS SDK 或 AWS SDK for PythonBoto3与 Step Functions API 进行交互。使用 Step Functions 客户端的 startexecution 方法，通过提供状态机的 Amazon 资源名称ARN和输入数据来调用 Step Functions 状态机。

在 Lambda 控制台上，创建您的 Lambda 函数，使用以下代码提供您之前创建的状态机的 ARN：

pythonimport jsonimport urllibparseimport boto3

print(Loading function)

STEPFUNCTIONARN = arnawsstatesuseast1stateMachineDMSS3postgresstepFunctionsstepfunctionsclient = boto3client(stepfunctions useast1)

def lambdahandler(event context) stepfunctionsclientstartexecution(stateMachineArn= STEPFUNCTIONARN input={} )

创建 S3 事件通知

通过 S3 事件通知，每当在指定的 S3 桶中创建或上传对象时，Lambda 函数将根据 S3 事件信息进行触发，并处理该事件。

要使用 Amazon S3 控制台启用 Lambda 通知，完成以下步骤：

如果您上传了大文件，使用 分段上传，并选择 s3ObjectCreatedCompleteMultipartUpload 事件。

有关更多信息，请参见 使用 Amazon S3 控制台启用和配置事件通知。

将 CSV 文件上传到 Amazon S3

要将数据文件上传到 Amazon S3，可以使用 Amazon S3 控制台或 AWS CLI。有关更多信息，请参见 上传对象。

![上传文件示例](https//d2908q01vomqb2cloudfrontnet/887309d048beef83ad3eabf2