如何在Aurora PostgreSQL插入X表数据后调用Step Functions并传参？

方案一：Aurora触发器 + Lambda 调用Step Functions #

这是最直接的实现方式，通过PostgreSQL触发器触发Lambda，再由Lambda启动Step Functions执行。

步骤： #

1. 配置IAM权限
   给Aurora集群关联的IAM角色添加以下权限：

   • lambda:InvokeFunction：允许调用指定Lambda函数
   • iam:PassRole：确保Aurora能传递角色权限调用Lambda

2. 启用aws_lambda扩展
   在Aurora PostgreSQL中执行SQL，启用Lambda集成扩展：

   CREATE EXTENSION IF NOT EXISTS aws_lambda;
   

3. 创建触发器函数
   编写PL/pgSQL函数，捕获X表的INSERT事件，将新数据传给Lambda：

   CREATE OR REPLACE FUNCTION trigger_invoke_sfn()
   RETURNS TRIGGER AS $$
   DECLARE
       payload JSON;
   BEGIN
       -- 将新插入的行转为JSON格式
       payload := row_to_json(NEW);
       
       -- 调用Lambda函数，替换为你的Lambda ARN
       PERFORM aws_lambda.invoke(
           'arn:aws:lambda:你的区域:账号ID:function:你的Lambda名称',
           json_build_object('new_record', payload)
       );
       
       RETURN NEW;
   END;
   $$ LANGUAGE plpgsql;
   

4. 绑定触发器到X表
   创建AFTER INSERT触发器，确保每次插入数据时执行上述函数：

   CREATE TRIGGER after_insert_x
   AFTER INSERT ON X
   FOR EACH ROW
   EXECUTE FUNCTION trigger_invoke_sfn();
   

5. 配置Lambda函数
   在Lambda中解析Aurora传来的数据，调用Step Functions的StartExecution API：

   import boto3
   import json
   
   sfn = boto3.client('stepfunctions')
   
   def lambda_handler(event, context):
       # 提取插入的新数据
       new_record = event['new_record']
       
       # 构造Step Functions执行输入
       sfn_input = json.dumps({'inserted_data': new_record})
       
       # 启动Step Functions执行，替换为你的状态机ARN
       response = sfn.start_execution(
           stateMachineArn='arn:aws:states:你的区域:账号ID:stateMachine:你的状态机名称',
           input=sfn_input
       )
       
       return {'execution_arn': response['executionArn']}
   

6. Step Functions触发目标Lambda
   在你的状态机中，添加Lambda任务节点，将Step Functions的执行结果作为输入传递给目标Lambda，完成后续逻辑。

方案二：Aurora CDC + Kinesis + Lambda + Step Functions #

适合高并发场景或不想直接在数据库层调用外部服务的需求，通过CDC捕获变更流，再流转到Step Functions。

步骤： #

1. 启用Aurora CDC功能
   开启Aurora PostgreSQL的逻辑复制，将X表的INSERT变更流推送到Kinesis Data Streams。

2. Kinesis触发Lambda
   配置Kinesis Data Streams的触发器，让Lambda接收CDC事件，过滤出INSERT类型的记录。

3. Lambda调用Step Functions
   解析CDC中的INSERT数据，逻辑同方案一的Lambda代码，调用Step Functions的StartExecution。

4. Step Functions触发目标Lambda
   同方案一，在状态机中配置Lambda任务，传递处理结果触发后续逻辑。

关键注意事项 #

• 权限校验：确保每个组件的IAM角色拥有对应操作权限（如Lambda调用Step Functions、Step Functions调用目标Lambda等）。
• 错误处理：触发器中添加异常捕获与日志记录，Step Functions配置重试策略，避免单次失败导致流程中断。
• 性能考量：直接用数据库触发器会增加插入操作的延迟，高并发场景优先选择CDC方案。
• 数据格式：确保JSON序列化/反序列化过程中数据结构一致，避免解析错误。

内容的提问来源于stack exchange，提问作者CHAHI Saad